From d5249ed9049f8d3bb06f6943a1a6d3b288f82d20 Mon Sep 17 00:00:00 2001 From: godotg Date: Sat, 20 Jan 2024 17:40:12 +0800 Subject: [PATCH] ref[provider]: refactor consistent hash load balancing --- .../balancer/ConsistentHashLoadBalancer.java | 68 +++++++++---------- .../com/zfoo/net/session/ISessionManager.java | 2 - .../com/zfoo/net/session/SessionManager.java | 13 ---- .../com/zfoo/net/util/ConsistentHash.java | 15 ++-- .../zfoo/protocol/collection/HashSetLong.java | 4 ++ 5 files changed, 41 insertions(+), 61 deletions(-) diff --git a/net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashLoadBalancer.java b/net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashLoadBalancer.java index 109260ba..87c33a94 100644 --- a/net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashLoadBalancer.java +++ b/net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashLoadBalancer.java @@ -19,7 +19,7 @@ import com.zfoo.net.util.ConsistentHash; import com.zfoo.net.util.FastTreeMapIntLong; import com.zfoo.net.util.HashUtils; import com.zfoo.protocol.ProtocolManager; -import com.zfoo.protocol.collection.CollectionUtils; +import com.zfoo.protocol.collection.HashSetLong; import com.zfoo.protocol.exception.RunException; import com.zfoo.protocol.model.Pair; import com.zfoo.protocol.registration.ProtocolModule; @@ -40,10 +40,19 @@ public class ConsistentHashLoadBalancer extends AbstractConsumerLoadBalancer { public static final ConsistentHashLoadBalancer INSTANCE = new ConsistentHashLoadBalancer(); - private volatile int lastClientSessionChangeId = 0; - private static final AtomicReferenceArray consistentHashMap = new AtomicReferenceArray<>(ProtocolManager.MAX_MODULE_NUM); + private static final AtomicReferenceArray consistentHashMap = new AtomicReferenceArray<>(ProtocolManager.MAX_MODULE_NUM); private static final int VIRTUAL_NODE_NUMS = 200; + public static class ConsistentCache { + public HashSetLong providerSids; + public FastTreeMapIntLong treeMap; + + public ConsistentCache(HashSetLong providerSids, FastTreeMapIntLong treeMap) { + this.providerSids = providerSids; + this.treeMap = treeMap; + } + } + public ConsistentHashLoadBalancer() { } @@ -64,21 +73,22 @@ public class ConsistentHashLoadBalancer extends AbstractConsumerLoadBalancer { return RandomLoadBalancer.getInstance().selectProvider(providers, packet, argument); } - updateConsistentHashMap(providers); - var module = ProtocolManager.moduleByProtocol(packet.getClass()); - var fastTreeMap = consistentHashMap.get(module.getId()); - if (fastTreeMap == null) { - fastTreeMap = updateModuleToConsistentHash(providers, module); + var consistentCache = consistentHashMap.get(module.getId()); + if (consistentCache == null) { + consistentCache = updateModuleToConsistentHash(providers, module); } - if (fastTreeMap == null) { - throw new RunException("ConsistentHashLoadBalancer [protocol:{}][argument:{}], no service provides the [module:{}]", packet.getClass(), argument, module); + var providerSids = consistentCache.providerSids; + // 一致性hash缓存不一致同样进行更新操作 + if (providerSids.size() != providers.size() || providers.stream().anyMatch(it -> !providerSids.contains(it.getSid()))) { + consistentCache = updateModuleToConsistentHash(providers, module); } - var nearestIndex = fastTreeMap.indexOfNearestCeilingKey(HashUtils.fnvHash(argument)); + var treeMap = consistentCache.treeMap; + var nearestIndex = treeMap.indexOfNearestCeilingKey(HashUtils.fnvHash(argument)); if (nearestIndex < 0) { throw new RunException("no service provides the [module:{}]", module); } - var sid = fastTreeMap.getByIndex(nearestIndex); + var sid = treeMap.getByIndex(nearestIndex); var session = NetContext.getSessionManager().getClientSession(sid); if (session == null) { throw new RunException("unknown no service provides the [module:{}]", module); @@ -86,25 +96,8 @@ public class ConsistentHashLoadBalancer extends AbstractConsumerLoadBalancer { return session; } - private void updateConsistentHashMap(List providers) { - // 如果更新时间不匹配，则更新到最新的服务提供者 - var currentClientSessionChangeId = NetContext.getSessionManager().getClientSessionChangeId(); - if (currentClientSessionChangeId != lastClientSessionChangeId) { - for (byte i = 0; i < ProtocolManager.MAX_MODULE_NUM; i++) { - var consistentHash = consistentHashMap.get(i); - if (consistentHash == null) { - continue; - } - var module = ProtocolManager.moduleByModuleId(i); - updateModuleToConsistentHash(providers, module); - } - lastClientSessionChangeId = currentClientSessionChangeId; - } - } - - @Nullable - private FastTreeMapIntLong updateModuleToConsistentHash(List providers, ProtocolModule module) { + private ConsistentCache updateModuleToConsistentHash(List providers, ProtocolModule module) { var sessionStringList = providers.stream() .map(session -> new Pair<>(session.getConsumerRegister().toString(), session.getSid())) .sorted((a, b) -> a.getKey().compareTo(b.getKey())) @@ -112,18 +105,21 @@ public class ConsistentHashLoadBalancer extends AbstractConsumerLoadBalancer { var consistentHash = new ConsistentHash<>(sessionStringList, VIRTUAL_NODE_NUMS); var virtualNodeTreeMap = consistentHash.getVirtualNodeTreeMap(); - if (CollectionUtils.isEmpty(virtualNodeTreeMap)) { - consistentHashMap.set(module.getId(), null); - return null; - } + var virtualTreeMap = new TreeMap(); for (var entry : virtualNodeTreeMap.entrySet()) { virtualTreeMap.put(entry.getKey(), entry.getValue().getValue()); } + + // 缓存服务提供者的sid + var sidSet = new HashSetLong(16); + providers.forEach(it -> sidSet.add(it.getSid())); // 使用更高性能的tree map var fastTreeMap = new FastTreeMapIntLong(virtualTreeMap); - consistentHashMap.set(module.getId(), fastTreeMap); - return fastTreeMap; + + var consistentCache = new ConsistentCache(sidSet, fastTreeMap); + consistentHashMap.set(module.getId(), consistentCache); + return consistentCache; } } diff --git a/net/src/main/java/com/zfoo/net/session/ISessionManager.java b/net/src/main/java/com/zfoo/net/session/ISessionManager.java index 59ba9810..dbff1121 100644 --- a/net/src/main/java/com/zfoo/net/session/ISessionManager.java +++ b/net/src/main/java/com/zfoo/net/session/ISessionManager.java @@ -39,6 +39,4 @@ public interface ISessionManager { int clientSessionSize(); - int getClientSessionChangeId(); - } diff --git a/net/src/main/java/com/zfoo/net/session/SessionManager.java b/net/src/main/java/com/zfoo/net/session/SessionManager.java index 22d501bb..6f9c9948 100644 --- a/net/src/main/java/com/zfoo/net/session/SessionManager.java +++ b/net/src/main/java/com/zfoo/net/session/SessionManager.java @@ -27,8 +27,6 @@ public class SessionManager implements ISessionManager { private static final Logger logger = LoggerFactory.getLogger(SessionManager.class); - private static final AtomicInteger CLIENT_ATOMIC = new AtomicInteger(0); - /** * EN: As a server, the Session is connected by other clients * CN: 作为服务器，被别的客户端连接的Session @@ -38,7 +36,6 @@ public class SessionManager implements ISessionManager { */ private final ConcurrentHashMapLongObject serverSessionMap = new ConcurrentHashMapLongObject<>(128); - /** * EN: As a client, connect to another server and save Sessions * CN: 作为客户端，连接别的服务器上后，保存下来的Session @@ -46,9 +43,6 @@ public class SessionManager implements ISessionManager { */ private final ConcurrentHashMapLongObject clientSessionMap = new ConcurrentHashMapLongObject<>(8); - private volatile int clientSessionChangeId = CLIENT_ATOMIC.incrementAndGet(); - - @Override public void addServerSession(Session session) { if (serverSessionMap.containsKey(session.getSid())) { @@ -91,7 +85,6 @@ public class SessionManager implements ISessionManager { return; } clientSessionMap.put(session.getSid(), session); - clientSessionChangeId = CLIENT_ATOMIC.incrementAndGet(); } @Override @@ -103,7 +96,6 @@ public class SessionManager implements ISessionManager { try (session) { clientSessionMap.remove(session.getSid()); } - clientSessionChangeId = CLIENT_ATOMIC.incrementAndGet(); } @Override @@ -121,9 +113,4 @@ public class SessionManager implements ISessionManager { return clientSessionMap.size(); } - @Override - public int getClientSessionChangeId() { - return clientSessionChangeId; - } - } diff --git a/net/src/main/java/com/zfoo/net/util/ConsistentHash.java b/net/src/main/java/com/zfoo/net/util/ConsistentHash.java index 375b0486..4cff91fb 100644 --- a/net/src/main/java/com/zfoo/net/util/ConsistentHash.java +++ b/net/src/main/java/com/zfoo/net/util/ConsistentHash.java @@ -41,18 +41,13 @@ public class ConsistentHash { this.realNodes.addAll(realNodes); this.virtualNodes = virtualNodes; - // 初始化 // 再添加虚拟节点，遍历LinkedList使用foreach循环效率会比较高 for (var realNode : realNodes) { - addNode(realNode); - } - } - - public void addNode(Pair realNode) { - for (var i = 0; i < this.virtualNodes; i++) { - var virtualNode = realNode.getKey().toString() + "&&VN" + i; - var hash = HashUtils.fnvHash(virtualNode); - virtualNodeTreeMap.put(hash, realNode); + for (var i = 0; i < this.virtualNodes; i++) { + var virtualNode = realNode.getKey().toString() + "&&VN" + i; + var hash = HashUtils.fnvHash(virtualNode); + virtualNodeTreeMap.put(hash, realNode); + } } } diff --git a/protocol/src/main/java/com/zfoo/protocol/collection/HashSetLong.java b/protocol/src/main/java/com/zfoo/protocol/collection/HashSetLong.java index 28ad4c4f..ad4c84bd 100644 --- a/protocol/src/main/java/com/zfoo/protocol/collection/HashSetLong.java +++ b/protocol/src/main/java/com/zfoo/protocol/collection/HashSetLong.java @@ -53,6 +53,10 @@ public class HashSetLong extends AbstractSet { return map.containsKey(o); } + public boolean contains(long key) { + return map.containsKey(key); + } + @Override public boolean add(Long e) { return map.put(e, Boolean.TRUE) == null;