动态调整映射关系的一致性哈希负载均衡算法详解

news2025/4/21 11:38:59

一、核心原理与设计要点

双重映射结构
一致性哈希负载均衡通过哈希环和槽动态分配实现双重映射关系：
• 哈希环构建：将节点（物理或虚拟）和数据键（Key）通过哈希函数（如MD5、CRC32）映射到固定范围的环形空间（如0~2³²），形成逻辑哈希环。

• 槽动态分配：将哈希环划分为固定数量的槽（Slot，如1024或16384个），每个槽对应哈希环上的一段范围，初始均匀分配到节点。运行时统计每个槽的请求量，通过动态规划或贪心算法重新分配槽到节点的映射，使各节点负载接近平均值。
动态调整流程
• 请求统计：周期性（如每分钟）记录每个槽的请求量，形成请求分布数组。

• 负载均衡目标：计算总请求量和节点数，得出目标平均负载（Avg = Total / 节点数）。

• 槽分段策略：将槽序列分割为连续段，使每段请求量与Avg的绝对差之和最小。例如，若当前段总和接近Avg则截断，剩余槽同理分配。

• 映射更新：调整后，同一槽的请求仍路由到同一节点，但槽与节点的映射关系随负载动态变化。
虚拟节点与权重适配
• 虚拟节点：每个物理节点对应多个虚拟节点（如100个），分散哈希环分布以缓解数据倾斜。

• 权重动态调整：根据节点性能差异，动态分配虚拟节点数量（如高性能节点分配更多虚拟节点），优化负载均衡。

二、Java代码实现

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicIntegerArray;

public class DynamicConsistentHash {
    private static final int SLOT_COUNT = 1024; // 槽总数
    private final TreeMap<Integer, String> ring = new TreeMap<>(); // 哈希环
    private final ConcurrentHashMap<String, List<Integer>> nodeSlots = new ConcurrentHashMap<>();
    private final AtomicIntegerArray slotRequests = new AtomicIntegerArray(SLOT_COUNT); // 槽请求统计

    // 初始化节点和虚拟节点
    public void initNodes(List<String> nodes, int virtualNodesPerNode) {
        nodes.forEach(node -> {
            List<Integer> slots = new ArrayList<>();
            for (int i = 0; i < virtualNodesPerNode; i++) {
                String virtualNode = node + "#VN" + i;
                int slot = Math.abs(virtualNode.hashCode()) % SLOT_COUNT;
                ring.put(slot, node);
                slots.add(slot);
            }
            nodeSlots.put(node, slots);
        });
    }

    // 动态调整槽映射（每1分钟触发）
    public void adjustSlots() {
        int total = 0;
        for (int i = 0; i < SLOT_COUNT; i++) total += slotRequests.get(i);
        double avg = total / (double) nodeSlots.size();

        // 按槽请求量降序排序
        List<Map.Entry<Integer, Integer>> sortedSlots = new ArrayList<>();
        for (int i = 0; i < SLOT_COUNT; i++) {
            sortedSlots.add(new AbstractMap.SimpleEntry<>(i, slotRequests.get(i)));
        }
        sortedSlots.sort((a, b) -> b.getValue() - a.getValue());

        // 贪心分配：高负载槽优先分配至低负载节点
        PriorityQueue<NodeLoad> loadQueue = new PriorityQueue<>();
        nodeSlots.keySet().forEach(node -> loadQueue.add(new NodeLoad(node, 0)));

        Map<String, List<Integer>> newMapping = new HashMap<>();
        for (Map.Entry<Integer, Integer> entry : sortedSlots) {
            NodeLoad node = loadQueue.poll();
            newMapping.computeIfAbsent(node.node, k -> new ArrayList<>()).add(entry.getKey());
            node.load += entry.getValue();
            loadQueue.add(node);
        }

        // 更新哈希环并重置统计
        newMapping.forEach((node, slots) -> slots.forEach(slot -> ring.put(slot, node)));
        slotRequests = new AtomicIntegerArray(SLOT_COUNT);
    }

    // 请求路由（记录槽请求量）
    public String route(String key) {
        int slot = Math.abs(key.hashCode()) % SLOT_COUNT;
        slotRequests.incrementAndGet(slot); // 统计请求量

        SortedMap<Integer, String> tailMap = ring.tailMap(slot);
        Integer targetSlot = tailMap.isEmpty() ? ring.firstKey() : tailMap.firstKey();
        return ring.get(targetSlot);
    }

    // 节点负载辅助类
    private static class NodeLoad implements Comparable<NodeLoad> {
        String node;
        int load;
        NodeLoad(String node, int load) { this.node = node; this.load = load; }
        @Override
        public int compareTo(NodeLoad o) { return Integer.compare(this.load, o.load); }
    }
}