本文目录导读：
在Java分布式系统中进行随机搜索或随机数据访问,核心挑战在于跨节点的随机性、一致性和性能，以下是几种常见的实现策略和代码示例，从简单到复杂。
核心概念
- 随机性：需要真随机或伪随机数生成器
- 分布式一致性：多节点都能独立生成相同或不同的随机结果
- 负载均衡：随机访问不应导致某些节点过载
- 数据分片：数据分布在多个节点上
实现策略
基于一致性哈希的随机路由
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.nio.ByteBuffer;
import java.util.*;
public class ConsistentHashRandomSearch<T> {
private final TreeMap<Long, T> ring = new TreeMap<>();
private final int numberOfReplicas;
private final MessageDigest md;
public ConsistentHashRandomSearch(int numberOfReplicas) throws NoSuchAlgorithmException {
this.numberOfReplicas = numberOfReplicas;
this.md = MessageDigest.getInstance("MD5");
}
public void addNode(T node) {
for (int i = 0; i < numberOfReplicas; i++) {
long hash = hash(node.toString() + ":" + i);
ring.put(hash, node);
}
}
public T getRandomNode() {
// 生成随机哈希值作为搜索起点
long randomHash = hash(UUID.randomUUID().toString());
Map.Entry<Long, T> entry = ring.ceilingEntry(randomHash);
if (entry == null) {
entry = ring.firstEntry();
}
return entry.getValue();
}
private long hash(String key) {
md.update(key.getBytes());
byte[] digest = md.digest();
return ByteBuffer.wrap(digest).getLong() & Long.MAX_VALUE;
}
}
使用场景：适合需要将请求均匀分布到不同数据分片的场景。
基于分片的随机采样
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;
public class ShardedRandomSearch<T> {
private final Map<Integer, List<T>> shards = new ConcurrentHashMap<>();
private final Random random = new Random();
private final int totalShards;
public ShardedRandomSearch(int totalShards) {
this.totalShards = totalShards;
for (int i = 0; i < totalShards; i++) {
shards.put(i, new CopyOnWriteArrayList<>());
}
}
// 添加数据到随机分片
public void addData(T data) {
int shardId = random.nextInt(totalShards);
shards.get(shardId).add(data);
}
// 随机搜索：先随机选分片，再从分片中随机选
public T randomSearch() {
if (shards.isEmpty()) return null;
// 1. 随机选择一个分片
int shardId = random.nextInt(totalShards);
List<T> shard = shards.get(shardId);
// 2. 从分片中随机选一个元素
if (shard.isEmpty()) {
// 如果该分片为空，尝试其他分片
for (int i = 0; i < totalShards; i++) {
shardId = (shardId + 1) % totalShards;
shard = shards.get(shardId);
if (!shard.isEmpty()) break;
}
}
return shard.get(random.nextInt(shard.size()));
}
// 批量随机搜索（不重复）
public List<T> batchRandomSearch(int count) {
List<T> results = new ArrayList<>();
Set<Integer> usedIndices = new HashSet<>();
while (results.size() < count) {
T item = randomSearch();
if (item != null && !usedIndices.contains(item.hashCode())) {
results.add(item);
usedIndices.add(item.hashCode());
}
}
return results;
}
}
分布式缓存随机访问
import redis.clients.jedis.*;
import java.util.*;
public class RedisRandomSearch {
private final JedisCluster jedisCluster;
private final Random random = new Random();
private final String[] nodes; // 所有Redis节点
public RedisRandomSearch(Set<HostAndPort> clusterNodes) {
this.jedisCluster = new JedisCluster(clusterNodes);
this.nodes = clusterNodes.stream()
.map(n -> n.getHost() + ":" + n.getPort())
.toArray(String[]::new);
}
// 存储数据时生成随机键
public void storeData(String key, String value) {
jedisCluster.set(key, value);
}
// 随机获取一条数据
public Map.Entry<String, String> randomGet() {
// 1. 随机选择一个节点
String node = nodes[random.nextInt(nodes.length)];
// 2. 使用 RANDOMKEY 命令
try (Jedis jedis = new Jedis(node)) {
String randomKey = jedis.randomKey();
if (randomKey != null) {
String value = jedisCluster.get(randomKey);
return new AbstractMap.SimpleEntry<>(randomKey, value);
}
}
return null;
}
// 扫描随机数据（大数据量）
public List<Map.Entry<String, String>> scanRandomData(int count) {
List<Map.Entry<String, String>> results = new ArrayList<>();
String cursor = "0";
ScanParams params = new ScanParams().count(100);
while (results.size() < count) {
ScanResult<String> scanResult = jedisCluster.scan(cursor, params);
for (String key : scanResult.getResult()) {
if (results.size() >= count) break;
String value = jedisCluster.get(key);
results.add(new AbstractMap.SimpleEntry<>(key, value));
}
cursor = scanResult.getCursor();
if (cursor.equals("0")) break;
}
return results;
}
}
基于权重的不均匀随机搜索
import java.util.*;
import java.util.concurrent.ThreadLocalRandom;
public class WeightedRandomSearch {
private final NavigableMap<Double, String> weightedMap = new TreeMap<>();
private double totalWeight = 0;
// 添加带权重的数据
public void addItem(String item, double weight) {
totalWeight += weight;
weightedMap.put(totalWeight, item);
}
// 根据权重随机选择
public String weightedRandom() {
double random = ThreadLocalRandom.current().nextDouble() * totalWeight;
Map.Entry<Double, String> entry = weightedMap.ceilingEntry(random);
return entry != null ? entry.getValue() : weightedMap.firstEntry().getValue();
}
// 分布式版本：从不同节点按权重获取
public static class DistributedWeightedSearch {
private final Map<String, WeightedRandomSearch> nodeSearches = new HashMap<>();
private final Map<String, Double> nodeWeights = new HashMap<>();
public void addNode(String nodeId, double weight) {
nodeSearches.put(nodeId, new WeightedRandomSearch());
nodeWeights.put(nodeId, weight);
}
public String distributedRandom() {
// 先随机选节点
double totalNodeWeight = nodeWeights.values().stream().mapToDouble(Double::doubleValue).sum();
double random = ThreadLocalRandom.current().nextDouble() * totalNodeWeight;
String selectedNode = null;
double cumulative = 0;
for (Map.Entry<String, Double> entry : nodeWeights.entrySet()) {
cumulative += entry.getValue();
if (random <= cumulative) {
selectedNode = entry.getKey();
break;
}
}
// 从选中节点随机获取
if (selectedNode != null) {
WeightedRandomSearch search = nodeSearches.get(selectedNode);
return search != null ? search.weightedRandom() : null;
}
return null;
}
}
}
基于流式处理的随机抽样
import java.util.*;
import java.util.concurrent.*;
public class ReservoirSampling<T> {
private final List<T> reservoir = new ArrayList<>();
private int count = 0;
private final int sampleSize;
public ReservoirSampling(int sampleSize) {
this.sampleSize = sampleSize;
}
// 蓄水池抽样算法
public void add(T item) {
count++;
if (count <= sampleSize) {
reservoir.add(item);
} else {
int index = ThreadLocalRandom.current().nextInt(count);
if (index < sampleSize) {
reservoir.set(index, item);
}
}
}
public List<T> getSample() {
return new ArrayList<>(reservoir);
}
// 分布式蓄水池抽样
public static class DistributedReservoir<T> {
private final Map<String, ReservoirSampling<T>> nodeSamples = new ConcurrentHashMap<>();
public void addItem(String node, T item) {
nodeSamples.computeIfAbsent(node, k -> new ReservoirSampling<>(100));
nodeSamples.get(node).add(item);
}
public List<T> getDistributedSample() {
// 从每个节点收集样本
List<T> combinedSample = new ArrayList<>();
for (ReservoirSampling<T> sampler : nodeSamples.values()) {
combinedSample.addAll(sampler.getSample());
}
// 对合并后的样本进行二次抽样
ReservoirSampling<T> finalSampler = new ReservoirSampling<>(100);
for (T item : combinedSample) {
finalSampler.add(item);
}
return finalSampler.getSample();
}
}
}
性能优化建议
-
预计算随机序列：
public class PrecomputedRandomSequence { private final List<Long> randomSequence; private int index = 0; public PrecomputedRandomSequence(int size) { randomSequence = new ArrayList<>(size); Random rnd = new Random(); for (int i = 0; i < size; i++) { randomSequence.add(rnd.nextLong()); } } public long nextRandom() { long value = randomSequence.get(index); index = (index + 1) % randomSequence.size(); return value; } } -
使用线程安全的随机数生成器：
// 推荐使用 ThreadLocalRandom 而非 Random ThreadLocalRandom.current().nextInt();
-
连接池管理： 对于数据库/缓存访问，务必使用连接池来减少开销。
完整示例：分布式随机用户推荐
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.*;
public class DistributedRandomRecommender {
private final Map<String, List<String>> userDatabase = new ConcurrentHashMap<>();
private final Random random = ThreadLocalRandom.current();
// 模拟分布式环境中的随机推荐
public List<String> getRandomRecommendations(String userId, int count) {
// 1. 从不同节点获取候选集
List<String> candidates = new ArrayList<>();
for (String node : userDatabase.keySet()) {
List<String> nodeData = userDatabase.get(node);
if (nodeData.size() > 0) {
// 从每个节点随机取一个子集
int sampleSize = Math.min(count, nodeData.size() / 10);
List<String> sample = new ArrayList<>(nodeData);
Collections.shuffle(sample, random);
candidates.addAll(sample.subList(0, sampleSize));
}
}
// 2. 对合并的候选集进行随机排序和选择
Collections.shuffle(candidates, random);
return candidates.stream()
.limit(count)
.collect(Collectors.toList());
}
// 基于用户历史行为加权随机
public List<String> getWeightedRecommendations(String userId, Map<String, Double> userHistory) {
WeightedRandomSearch weightedSearch = new WeightedRandomSearch();
// 根据用户历史行为计算权重
for (Map.Entry<String, Double> history : userHistory.entrySet()) {
weightedSearch.addItem(history.getKey(), history.getValue());
}
List<String> recommendations = new ArrayList<>();
for (int i = 0; i < 10; i++) {
String item = weightedSearch.weightedRandom();
if (item != null) {
recommendations.add(item);
}
}
return recommendations;
}
}
选择哪种策略取决于：
- 数据规模：小数据用简单随机，大数据用分片随机
- 分布均匀性要求：一致性哈希适合均匀分布，蓄水池抽样适合流式数据
- 性能要求：预计算序列提高性能，加权随机适合业务场景
- 一致性要求：强一致性用分布式锁，最终一致性用乐观锁
实际生产环境建议结合多种策略,并进行充分的压测和调优。
