Java分布式数据随机搜索等怎么随机

wen java案例 2026-07-05 265

本文目录导读：

Java分布式数据随机搜索等怎么随机

核心概念
实现策略
性能优化建议
完整示例：分布式随机用户推荐

在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;
    }
}

选择哪种策略取决于：