From 82782d71eab1f51d08f60300ff9559c835ea2743 Mon Sep 17 00:00:00 2001 From: godotg Date: Sat, 13 Jan 2024 14:07:57 +0800 Subject: [PATCH] ref[consumer]: simplify consumer configuration and support repeated consumption of the same interface by different service providers --- .../java/com/zfoo/net/consumer/Consumer.java | 93 ++++++++++++------- .../java/com/zfoo/net/consumer/IConsumer.java | 8 +- .../AbstractConsumerLoadBalancer.java | 41 -------- .../ConsistentHashConsumerLoadBalancer.java | 17 ++-- ...stentHashOfMemoryConsumerLoadBalancer.java | 9 +- .../balancer/IConsumerLoadBalancer.java | 5 +- .../balancer/RandomConsumerLoadBalancer.java | 15 +-- .../consumer/registry/ZookeeperRegistry.java | 73 ++++++++------- .../zfoo/net/handler/GatewayRouteHandler.java | 11 +-- 9 files changed, 129 insertions(+), 143 deletions(-) diff --git a/net/src/main/java/com/zfoo/net/consumer/Consumer.java b/net/src/main/java/com/zfoo/net/consumer/Consumer.java index f785f811..533516ce 100644 --- a/net/src/main/java/com/zfoo/net/consumer/Consumer.java +++ b/net/src/main/java/com/zfoo/net/consumer/Consumer.java @@ -18,6 +18,7 @@ import com.zfoo.net.consumer.balancer.AbstractConsumerLoadBalancer; import com.zfoo.net.consumer.balancer.IConsumerLoadBalancer; import com.zfoo.net.packet.common.Error; import com.zfoo.net.router.Router; +import com.zfoo.net.router.SignalBridge; import com.zfoo.net.router.answer.AsyncAnswer; import com.zfoo.net.router.answer.SyncAnswer; import com.zfoo.net.router.attachment.NoAnswerAttachment; @@ -25,12 +26,11 @@ import com.zfoo.net.router.attachment.SignalAttachment; import com.zfoo.net.router.exception.ErrorResponseException; import com.zfoo.net.router.exception.NetTimeOutException; import com.zfoo.net.router.exception.UnexpectedProtocolException; -import com.zfoo.net.router.SignalBridge; import com.zfoo.net.session.Session; import com.zfoo.net.task.TaskBus; import com.zfoo.protocol.ProtocolManager; import com.zfoo.protocol.collection.CollectionUtils; -import com.zfoo.protocol.registration.ProtocolModule; +import com.zfoo.protocol.exception.RunException; import com.zfoo.protocol.util.JsonUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -53,6 +53,7 @@ public class Consumer implements IConsumer { private static final Logger logger = LoggerFactory.getLogger(Consumer.class); + // consumer|provider -> LoadBalancer private final Map consumerLoadBalancerMap = new HashMap<>(); @Override @@ -69,50 +70,74 @@ public class Consumer implements IConsumer { } - public List getSessionsByModule(ProtocolModule module) { + // find all session that can process interface/packet of protocolModule + @Override + public List findProviders(Object packet) { + var protocolModule = ProtocolManager.moduleByProtocol(packet.getClass()); var list = new ArrayList(); - NetContext.getSessionManager().forEachClientSession(new java.util.function.Consumer() { - @Override - public void accept(Session session) { - if (session.getConsumerAttribute() == null || session.getConsumerAttribute().getProviderConfig() == null) { - return; - } - var providerConfig = session.getConsumerAttribute().getProviderConfig(); - if (providerConfig.getProviders().stream().anyMatch(it -> it.getProtocolModule().equals(module))) { - list.add(session); - } + NetContext.getSessionManager().forEachClientSession(session -> { + var consumerAttribute = session.getConsumerAttribute(); + if (consumerAttribute == null) { + return; } + var providerConfig = consumerAttribute.getProviderConfig(); + if (providerConfig == null) { + return; + } + var providers = providerConfig.getProviders(); + if (providers == null) { + return; + } + if (providers.stream().noneMatch(it -> it.getProtocolModule().equals(protocolModule.getName()))) { + return; + } + list.add(session); }); + if (CollectionUtils.isEmpty(list)) { + throw new RunException("[protocol:{}] has no service that provides the [module:{}]", packet.getClass().getSimpleName(), protocolModule); + } return list; } - public void getProvider(Object packet) { - var protocolMModule = ProtocolManager.moduleByProtocol(packet.getClass()); - - } - + // Select a consumer loadBalancer @Override - public IConsumerLoadBalancer loadBalancer(ProtocolModule protocolModule) { - return consumerLoadBalancerMap.get(protocolModule); + public IConsumerLoadBalancer selectLoadBalancer(List providers, Object packet) { + // select first consumer loadBalancer + // 不同的服务提供者可能会提供同一个接口，消费者可能同时消费了这些提供了同一个接口的服务提供者，取第一个消费者的loadBalancer + IConsumerLoadBalancer loadBalancer = null; + for (var providerSession : providers) { + for (var provider : providerSession.getConsumerAttribute().getProviderConfig().getProviders()) { + if (consumerLoadBalancerMap.containsKey(provider.getProvider())) { + loadBalancer = consumerLoadBalancerMap.get(provider.getProvider()); + break; + } + } + if (loadBalancer != null) { + break; + } + } + if (loadBalancer == null) { + var protocolModule = ProtocolManager.moduleByProtocol(packet.getClass()); + throw new RunException("[protocol:{}] can not find any loadBalancer for the [module:{}]", packet.getClass().getSimpleName(), protocolModule); + } + return loadBalancer; } + @Override public void send(Object packet, Object argument) { - try { - var loadBalancer = loadBalancer(ProtocolManager.moduleByProtocol(packet.getClass())); - var session = loadBalancer.loadBalancer(packet, argument); - var taskExecutorHash = TaskBus.calTaskExecutorHash(argument); - NetContext.getRouter().send(session, packet, NoAnswerAttachment.valueOf(taskExecutorHash)); - } catch (Throwable t) { - logger.error("consumer unknown exception", t); - } + var providers = findProviders(packet); + var loadBalancer = selectLoadBalancer(providers, packet); + var session = loadBalancer.selectProvider(providers, packet, argument); + var taskExecutorHash = TaskBus.calTaskExecutorHash(argument); + NetContext.getRouter().send(session, packet, NoAnswerAttachment.valueOf(taskExecutorHash)); } @Override public SyncAnswer syncAsk(Object packet, Class answerClass, Object argument) throws Exception { - var loadBalancer = loadBalancer(ProtocolManager.moduleByProtocol(packet.getClass())); - var session = loadBalancer.loadBalancer(packet, argument); - + var providers = findProviders(packet); + var loadBalancer = selectLoadBalancer(providers, packet); + var session = loadBalancer.selectProvider(providers, packet, argument); // 下面的代码逻辑同Router的syncAsk，如果修改的话，记得一起修改 var clientSignalAttachment = new SignalAttachment(); @@ -150,8 +175,10 @@ public class Consumer implements IConsumer { @Override public AsyncAnswer asyncAsk(Object packet, Class answerClass, Object argument) { - var loadBalancer = loadBalancer(ProtocolManager.moduleByProtocol(packet.getClass())); - var session = loadBalancer.loadBalancer(packet, argument); + var providers = findProviders(packet); + var loadBalancer = selectLoadBalancer(providers, packet); + var session = loadBalancer.selectProvider(providers, packet, argument); + var asyncAnswer = NetContext.getRouter().asyncAsk(session, packet, answerClass, argument); // load balancer之前调用 diff --git a/net/src/main/java/com/zfoo/net/consumer/IConsumer.java b/net/src/main/java/com/zfoo/net/consumer/IConsumer.java index 1da5f3a4..7302d912 100644 --- a/net/src/main/java/com/zfoo/net/consumer/IConsumer.java +++ b/net/src/main/java/com/zfoo/net/consumer/IConsumer.java @@ -16,9 +16,11 @@ package com.zfoo.net.consumer; import com.zfoo.net.consumer.balancer.IConsumerLoadBalancer; import com.zfoo.net.router.answer.AsyncAnswer; import com.zfoo.net.router.answer.SyncAnswer; -import com.zfoo.protocol.registration.ProtocolModule; +import com.zfoo.net.session.Session; import org.springframework.lang.Nullable; +import java.util.List; + /** * @author godotg */ @@ -26,7 +28,9 @@ public interface IConsumer { void init(); - IConsumerLoadBalancer loadBalancer(ProtocolModule protocolModule); + List findProviders(Object packet); + + IConsumerLoadBalancer selectLoadBalancer(List providers, Object packet); /** * 直接发送，不需要任何返回值 diff --git a/net/src/main/java/com/zfoo/net/consumer/balancer/AbstractConsumerLoadBalancer.java b/net/src/main/java/com/zfoo/net/consumer/balancer/AbstractConsumerLoadBalancer.java index d1e91232..d1cf3b45 100644 --- a/net/src/main/java/com/zfoo/net/consumer/balancer/AbstractConsumerLoadBalancer.java +++ b/net/src/main/java/com/zfoo/net/consumer/balancer/AbstractConsumerLoadBalancer.java @@ -13,18 +13,8 @@ package com.zfoo.net.consumer.balancer; -import com.zfoo.net.NetContext; -import com.zfoo.net.consumer.registry.RegisterVO; -import com.zfoo.net.session.Session; -import com.zfoo.protocol.ProtocolManager; -import com.zfoo.protocol.registration.ProtocolModule; import com.zfoo.protocol.util.StringUtils; -import java.util.ArrayList; -import java.util.List; -import java.util.Objects; -import java.util.function.Consumer; - /** * @author godotg */ @@ -48,35 +38,4 @@ public abstract class AbstractConsumerLoadBalancer implements IConsumerLoadBalan return balancer; } - public List getSessionsByModule(ProtocolModule module) { - var list = new ArrayList(); - NetContext.getSessionManager().forEachClientSession(new Consumer() { - @Override - public void accept(Session session) { - if (session.getConsumerAttribute() == null || session.getConsumerAttribute().getProviderConfig() == null) { - return; - } - var providerConfig = session.getConsumerAttribute().getProviderConfig(); - if (providerConfig.getProviders().stream().anyMatch(it -> it.getProtocolModule().equals(module))) { - list.add(session); - } - } - }); - return list; - } - - public boolean sessionHasModule(Session session, Object packet) { - var consumerAttribute = session.getConsumerAttribute(); - if (Objects.isNull(consumerAttribute)) { - return false; - } - - var registerVO = (RegisterVO) consumerAttribute; - if (Objects.isNull(registerVO.getProviderConfig())) { - return false; - } - - var module = ProtocolManager.moduleByProtocol(packet.getClass()); - return registerVO.getProviderConfig().getProviders().contains(module); - } } diff --git a/net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashConsumerLoadBalancer.java b/net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashConsumerLoadBalancer.java index 65ee46db..eb1ad2eb 100644 --- a/net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashConsumerLoadBalancer.java +++ b/net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashConsumerLoadBalancer.java @@ -25,6 +25,7 @@ import com.zfoo.protocol.model.Pair; import com.zfoo.protocol.registration.ProtocolModule; import org.springframework.lang.Nullable; +import java.util.List; import java.util.TreeMap; import java.util.concurrent.atomic.AtomicReferenceArray; @@ -58,17 +59,17 @@ public class ConsistentHashConsumerLoadBalancer extends AbstractConsumerLoadBala * @return 调用的session */ @Override - public Session loadBalancer(Object packet, Object argument) { + public Session selectProvider(List providers, Object packet, Object argument) { if (argument == null) { - return RandomConsumerLoadBalancer.getInstance().loadBalancer(packet, argument); + return RandomConsumerLoadBalancer.getInstance().selectProvider(providers, packet, argument); } - updateConsistentHashMap(); + updateConsistentHashMap(providers); var module = ProtocolManager.moduleByProtocol(packet.getClass()); var fastTreeMap = consistentHashMap.get(module.getId()); if (fastTreeMap == null) { - fastTreeMap = updateModuleToConsistentHash(module); + fastTreeMap = updateModuleToConsistentHash(providers, module); } if (fastTreeMap == null) { throw new RunException("ConsistentHashLoadBalancer [protocol:{}][argument:{}], no service provides the [module:{}]", packet.getClass(), argument, module); @@ -85,7 +86,7 @@ public class ConsistentHashConsumerLoadBalancer extends AbstractConsumerLoadBala return session; } - private void updateConsistentHashMap() { + private void updateConsistentHashMap(List providers) { // 如果更新时间不匹配，则更新到最新的服务提供者 var currentClientSessionChangeId = NetContext.getSessionManager().getClientSessionChangeId(); if (currentClientSessionChangeId != lastClientSessionChangeId) { @@ -95,7 +96,7 @@ public class ConsistentHashConsumerLoadBalancer extends AbstractConsumerLoadBala continue; } var module = ProtocolManager.moduleByModuleId(i); - updateModuleToConsistentHash(module); + updateModuleToConsistentHash(providers, module); } lastClientSessionChangeId = currentClientSessionChangeId; } @@ -103,8 +104,8 @@ public class ConsistentHashConsumerLoadBalancer extends AbstractConsumerLoadBala @Nullable - private FastTreeMapIntLong updateModuleToConsistentHash(ProtocolModule module) { - var sessionStringList = getSessionsByModule(module).stream() + private FastTreeMapIntLong updateModuleToConsistentHash(List providers, ProtocolModule module) { + var sessionStringList = providers.stream() .map(session -> new Pair<>(session.getConsumerAttribute().toString(), session.getSid())) .sorted((a, b) -> a.getKey().compareTo(b.getKey())) .toList(); diff --git a/net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashOfMemoryConsumerLoadBalancer.java b/net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashOfMemoryConsumerLoadBalancer.java index 8e8fb365..fd83227e 100644 --- a/net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashOfMemoryConsumerLoadBalancer.java +++ b/net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashOfMemoryConsumerLoadBalancer.java @@ -2,7 +2,6 @@ package com.zfoo.net.consumer.balancer; import com.zfoo.net.NetContext; import com.zfoo.net.session.Session; -import org.apache.curator.shaded.com.google.common.collect.Lists; import org.apache.curator.shaded.com.google.common.util.concurrent.AtomicLongMap; import java.util.List; @@ -31,20 +30,20 @@ public class ConsistentHashOfMemoryConsumerLoadBalancer extends ConsistentHashCo } @Override - public Session loadBalancer(Object packet, Object argument) { + public Session selectProvider(List providers, Object packet, Object argument) { if (argument instanceof Long) { long sid = uid2sidMap.get((Long) argument); if (sid > 0L) { Session memorySession = NetContext.getSessionManager().getClientSession(sid); - if (null != memorySession){ + if (null != memorySession) { return memorySession; - }else { + } else { uid2sidMap.remove((Long) argument); } } } - Session loadBalancer = super.loadBalancer(packet, argument); + Session loadBalancer = super.selectProvider(providers, packet, argument); if (argument instanceof Long){ uid2sidMap.put((Long) argument, loadBalancer.getSid()); } diff --git a/net/src/main/java/com/zfoo/net/consumer/balancer/IConsumerLoadBalancer.java b/net/src/main/java/com/zfoo/net/consumer/balancer/IConsumerLoadBalancer.java index c343a82a..02a5db8d 100644 --- a/net/src/main/java/com/zfoo/net/consumer/balancer/IConsumerLoadBalancer.java +++ b/net/src/main/java/com/zfoo/net/consumer/balancer/IConsumerLoadBalancer.java @@ -17,19 +17,22 @@ import com.zfoo.net.router.attachment.SignalAttachment; import com.zfoo.net.session.Session; import org.springframework.lang.Nullable; +import java.util.List; + /** * @author godotg */ public interface IConsumerLoadBalancer { /** + * Select a service provider that can provide interface/packet services * 只有一致性hash会使用这个argument参数，如果在一致性hash没有传入argument默认使用随机负载均衡 * * @param packet 请求包 * @param argument 计算参数 * @return 一个服务提供者的session */ - Session loadBalancer(Object packet, @Nullable Object argument); + Session selectProvider(List providers, Object packet, @Nullable Object argument); default void beforeLoadBalancer(Session session, Object packet, SignalAttachment attachment) { } diff --git a/net/src/main/java/com/zfoo/net/consumer/balancer/RandomConsumerLoadBalancer.java b/net/src/main/java/com/zfoo/net/consumer/balancer/RandomConsumerLoadBalancer.java index a781bbf1..26efc40b 100644 --- a/net/src/main/java/com/zfoo/net/consumer/balancer/RandomConsumerLoadBalancer.java +++ b/net/src/main/java/com/zfoo/net/consumer/balancer/RandomConsumerLoadBalancer.java @@ -14,10 +14,10 @@ package com.zfoo.net.consumer.balancer; import com.zfoo.net.session.Session; -import com.zfoo.protocol.ProtocolManager; -import com.zfoo.protocol.exception.RunException; import com.zfoo.protocol.util.RandomUtils; +import java.util.List; + /** * 随机负载均衡器，任选服务提供者的其中之一 * @@ -35,15 +35,8 @@ public class RandomConsumerLoadBalancer extends AbstractConsumerLoadBalancer { } @Override - public Session loadBalancer(Object packet, Object argument) { - var module = ProtocolManager.moduleByProtocol(packet.getClass()); - var sessions = getSessionsByModule(module); - - if (sessions.isEmpty()) { - throw new RunException("RandomConsumerLoadBalancer [protocol:{}][argument:{}], no service provides the [module:{}]", packet.getClass(), argument, module); - } - - return RandomUtils.randomEle(sessions); + public Session selectProvider(List providers, Object packet, Object argument) { + return RandomUtils.randomEle(providers); } } diff --git a/net/src/main/java/com/zfoo/net/consumer/registry/ZookeeperRegistry.java b/net/src/main/java/com/zfoo/net/consumer/registry/ZookeeperRegistry.java index 5bd723f2..2bba542f 100644 --- a/net/src/main/java/com/zfoo/net/consumer/registry/ZookeeperRegistry.java +++ b/net/src/main/java/com/zfoo/net/consumer/registry/ZookeeperRegistry.java @@ -432,7 +432,7 @@ public class ZookeeperRegistry implements IRegistry { return; } - executor.execute(() -> doCheckConsumer()); + executor.execute(ThreadUtils.safeRunnable(() -> doCheckConsumer())); } /** @@ -450,17 +450,17 @@ public class ZookeeperRegistry implements IRegistry { for (var providerCache : providerHashConsumerSet) { // 先排除已经启动的consumer - // getClientSessionMap var consumerClientList = new ArrayList(); NetContext.getSessionManager().forEachClientSession(new Consumer() { - @Override - public void accept(Session session) { - if (session.getConsumerAttribute() != null && session.getConsumerAttribute().equals(providerCache)) { - consumerClientList.add(session); - } - } - }); + @Override + public void accept(Session session) { + if (session.getConsumerAttribute() != null && session.getConsumerAttribute().equals(providerCache)) { + consumerClientList.add(session); + } + } + }); + // consumerClientList大于等于1，说明消费者连接成功了 if (consumerClientList.size() == 1) { var consumer = consumerClientList.get(0); if (SessionUtils.isActive(consumer)) { @@ -476,41 +476,44 @@ public class ZookeeperRegistry implements IRegistry { continue; } - // 自己作为消费者，要创建一个TcpClient去连接服务提供者 - var client = new TcpClient(HostAndPort.valueOf(providerCache.getProviderConfig().getAddress())); - var session = client.start(); + try { + // 自己作为消费者，要创建一个TcpClient去连接服务提供者 + var client = new TcpClient(HostAndPort.valueOf(providerCache.getProviderConfig().getAddress())); + var session = client.start(); + if (session == null) { + recheckFlag = true; + continue; + } - // 自己作为消费者，使用TcpClient连接服务提供者不成功 - if (Objects.isNull(session)) { - logger.error("[consumer:{}] failed to start, wait [{}] seconds to recheck consumer", providerCache, RETRY_SECONDS); - recheckFlag = true; - } else { - // 连接上了服务提供者 session.setConsumerAttribute(providerCache); EventBus.post(ConsumerStartEvent.valueOf(providerCache, session)); - - try { - var localRegisterVO = NetContext.getConfigManager().getLocalConfig().toLocalRegisterVO(); - var path = CONSUMER_ROOT_PATH + StringUtils.SLASH + localRegisterVO.toConsumerString(); - var stat = curator.checkExists().forPath(path); - if (Objects.isNull(stat)) { - curator.create() - .withMode(CreateMode.EPHEMERAL) - .forPath(path); - } else { - curator.setData().forPath(path); - } - - } catch (Exception e) { - // 因为并不关心consumer的状态，这种失败只需要记录一个错误日志就可以了 - logger.error("consumer writing to Zookeeper failed", e); - } + } catch (Throwable t) { + logger.error("[consumer:{}] failed to start, wait [{}] seconds to recheck consumer", providerCache, RETRY_SECONDS, t); + recheckFlag = true; } } + // 将自己的消费者消息写到 /consumer 的临时节点下 + var localRegisterVO = NetContext.getConfigManager().getLocalConfig().toLocalRegisterVO(); + var path = CONSUMER_ROOT_PATH + StringUtils.SLASH + localRegisterVO.toConsumerString(); + try { + var stat = curator.checkExists().forPath(path); + if (Objects.isNull(stat)) { + curator.create() + .withMode(CreateMode.EPHEMERAL) + .forPath(path); + } else { + curator.setData().forPath(path); + } + } catch (Exception e) { + logger.error("consumer:[{}] writing to Zookeeper failed", path, e); + recheckFlag = true; + } + if (recheckFlag) { SchedulerBus.schedule(() -> checkConsumer(), RETRY_SECONDS, TimeUnit.SECONDS); } + } /** diff --git a/net/src/main/java/com/zfoo/net/handler/GatewayRouteHandler.java b/net/src/main/java/com/zfoo/net/handler/GatewayRouteHandler.java index 2bb1ee4a..b215bec1 100644 --- a/net/src/main/java/com/zfoo/net/handler/GatewayRouteHandler.java +++ b/net/src/main/java/com/zfoo/net/handler/GatewayRouteHandler.java @@ -15,8 +15,6 @@ package com.zfoo.net.handler; import com.zfoo.event.manager.EventBus; import com.zfoo.net.NetContext; -import com.zfoo.net.consumer.balancer.ConsistentHashConsumerLoadBalancer; -import com.zfoo.net.consumer.balancer.IConsumerLoadBalancer; import com.zfoo.net.core.gateway.IGatewayLoadBalancer; import com.zfoo.net.core.gateway.model.GatewaySessionInactiveEvent; import com.zfoo.net.packet.DecodedPacketInfo; @@ -27,7 +25,6 @@ import com.zfoo.net.router.attachment.GatewayAttachment; import com.zfoo.net.router.attachment.SignalAttachment; import com.zfoo.net.session.Session; import com.zfoo.net.util.SessionUtils; -import com.zfoo.protocol.ProtocolManager; import com.zfoo.protocol.util.JsonUtils; import com.zfoo.protocol.util.StringUtils; import com.zfoo.scheduler.util.TimeUtils; @@ -117,10 +114,10 @@ public class GatewayRouteHandler extends ServerRouteHandler { private void forwardingPacket(Object packet, Object attachment, Object argument) { try { // 网关统一用 moduleid uid 获取 session - var loadBalancer = NetContext.getConsumer().loadBalancer(ProtocolManager.moduleByProtocol(packet.getClass())); - Session consumerSession = loadBalancer.loadBalancer(packet, argument); -// var consumerSession = ConsistentHashConsumerLoadBalancer.getInstance().loadBalancer(packet, argument); - NetContext.getRouter().send(consumerSession, packet, attachment); + var providers = NetContext.getConsumer().findProviders(packet); + var loadBalancer = NetContext.getConsumer().selectLoadBalancer(providers, packet); + var providerSession = loadBalancer.selectProvider(providers, packet, argument); + NetContext.getRouter().send(providerSession, packet, attachment); } catch (Exception e) { logger.error("An exception occurred at the gateway", e); } catch (Throwable t) {