当前位置：首页 > news >正文

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

news 来源：原创 2025/4/30 13:31:46

一、核心原理与设计要点

双重映射结构
一致性哈希负载均衡通过哈希环和槽动态分配实现双重映射关系：
• 哈希环构建：将节点（物理或虚拟）和数据键（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; }@Overridepublic int compareTo(NodeLoad o) { return Integer.compare(this.load, o.load); }}
}