From 02ad965755cd3a02f612a6c22e404e531880a33e Mon Sep 17 00:00:00 2001 From: godotg Date: Sat, 13 Jan 2024 11:47:13 +0800 Subject: [PATCH] ref[rpc]: refactor service provider --- .../com/zfoo/net/config/ConfigManager.java | 4 +-- .../zfoo/net/config/model/ProviderConfig.java | 3 ++ .../zfoo/net/config/model/ProviderModule.java | 2 -- .../java/com/zfoo/net/consumer/Consumer.java | 33 +++++++++++++++++-- .../net/consumer/registry/RegisterVO.java | 17 ++++------ .../consumer/registry/ZookeeperRegistry.java | 4 +-- .../com/zfoo/net/config/RegistryTest.java | 6 ++-- 7 files changed, 46 insertions(+), 23 deletions(-) diff --git a/net/src/main/java/com/zfoo/net/config/ConfigManager.java b/net/src/main/java/com/zfoo/net/config/ConfigManager.java index 10b1a2cc..8043ca81 100644 --- a/net/src/main/java/com/zfoo/net/config/ConfigManager.java +++ b/net/src/main/java/com/zfoo/net/config/ConfigManager.java @@ -66,7 +66,7 @@ public class ConfigManager implements IConfigManager { var provider = providerModule.getProvider(); var protocolModule = ProtocolManager.moduleByModuleName(provider); AssertionUtils.isTrue(protocolModule != null, "provider:[{}] does not exist in the protocol manager", provider); - AssertionUtils.isTrue(providerSet.add(provider), "provider:[{}] plicate Consumption Agreement module [provider:{}]", protocolModuleName, provider); + AssertionUtils.isTrue(providerSet.add(provider), "provider:[{}] has duplicate provider name module [provider:{}]", provider, protocolModule); } } @@ -77,7 +77,7 @@ public class ConfigManager implements IConfigManager { for (var consumerModule : consumerConfig.getConsumers()) { // 提供的接口实现 提供者名 var consumer = consumerModule.getConsumer(); - AssertionUtils.isTrue(consumerSet.add(consumer), "服务消费者[name:{}]重复消费了协议模块[consumer:{}]", protocolModuleName, consumer); + AssertionUtils.isTrue(consumerSet.add(consumer), "consumer:[{}] has duplicate consumer module", consumer); } } diff --git a/net/src/main/java/com/zfoo/net/config/model/ProviderConfig.java b/net/src/main/java/com/zfoo/net/config/model/ProviderConfig.java index 1fe26ea1..a31960cf 100644 --- a/net/src/main/java/com/zfoo/net/config/model/ProviderConfig.java +++ b/net/src/main/java/com/zfoo/net/config/model/ProviderConfig.java @@ -29,6 +29,9 @@ public class ProviderConfig { private String thread; + /** + * If no address is configured, the default address generated by localHostAndPortOrDefault() is used as the server address. + */ private String address; private List providers; diff --git a/net/src/main/java/com/zfoo/net/config/model/ProviderModule.java b/net/src/main/java/com/zfoo/net/config/model/ProviderModule.java index 49d2d431..98d7c571 100644 --- a/net/src/main/java/com/zfoo/net/config/model/ProviderModule.java +++ b/net/src/main/java/com/zfoo/net/config/model/ProviderModule.java @@ -12,8 +12,6 @@ package com.zfoo.net.config.model; -import com.zfoo.protocol.registration.ProtocolModule; - import java.util.Objects; /** 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 412f452f..f785f811 100644 --- a/net/src/main/java/com/zfoo/net/consumer/Consumer.java +++ b/net/src/main/java/com/zfoo/net/consumer/Consumer.java @@ -26,6 +26,7 @@ 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; @@ -34,7 +35,9 @@ import com.zfoo.protocol.util.JsonUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.ArrayList; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -50,7 +53,7 @@ public class Consumer implements IConsumer { private static final Logger logger = LoggerFactory.getLogger(Consumer.class); - private final Map consumerLoadBalancerMap = new HashMap<>(); + private final Map consumerLoadBalancerMap = new HashMap<>(); @Override public void init() { @@ -60,10 +63,34 @@ public class Consumer implements IConsumer { } var consumers = consumerConfig.getConsumers(); for (var consumer : consumers) { - consumerLoadBalancerMap.put(consumer.getProtocolModule(), AbstractConsumerLoadBalancer.valueOf(consumer.getLoadBalancer())); + var loadBalancer = AbstractConsumerLoadBalancer.valueOf(consumer.getLoadBalancer()); + consumerLoadBalancerMap.put(consumer.getConsumer(), loadBalancer); } } + + public List getSessionsByModule(ProtocolModule module) { + 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); + } + } + }); + return list; + } + + public void getProvider(Object packet) { + var protocolMModule = ProtocolManager.moduleByProtocol(packet.getClass()); + + } + @Override public IConsumerLoadBalancer loadBalancer(ProtocolModule protocolModule) { return consumerLoadBalancerMap.get(protocolModule); @@ -77,7 +104,7 @@ public class Consumer implements IConsumer { var taskExecutorHash = TaskBus.calTaskExecutorHash(argument); NetContext.getRouter().send(session, packet, NoAnswerAttachment.valueOf(taskExecutorHash)); } catch (Throwable t) { - logger.error("consumer发送未知异常", t); + logger.error("consumer unknown exception", t); } } diff --git a/net/src/main/java/com/zfoo/net/consumer/registry/RegisterVO.java b/net/src/main/java/com/zfoo/net/consumer/registry/RegisterVO.java index b97b2182..4892289c 100644 --- a/net/src/main/java/com/zfoo/net/consumer/registry/RegisterVO.java +++ b/net/src/main/java/com/zfoo/net/consumer/registry/RegisterVO.java @@ -99,14 +99,13 @@ public class RegisterVO { private static List parseProviderModules(String str) { - var moduleSplits = StringUtils.substringBeforeLast( - StringUtils.substringAfterFirst(str, StringUtils.LEFT_SQUARE_BRACKET) - , StringUtils.RIGHT_SQUARE_BRACKET).split(StringUtils.COMMA); - + str = StringUtils.substringAfterFirst(str, StringUtils.LEFT_SQUARE_BRACKET); + str = StringUtils.substringBeforeLast(str, StringUtils.RIGHT_SQUARE_BRACKET); + var moduleSplits = str.split(StringUtils.COMMA); var modules = Arrays.stream(moduleSplits) .map(it -> it.trim()) .map(it -> it.split(StringUtils.HYPHEN)) - .map(it -> new ProviderModule(new ProtocolModule(Byte.parseByte(it[0]), it[1]), it[2])) + .map(it -> new ProviderModule(it[0], it[1])) .toList(); return modules; } @@ -129,9 +128,7 @@ public class RegisterVO { } public String toConsumerString() { - return this + - StringUtils.SPACE + StringUtils.VERTICAL_BAR + StringUtils.SPACE + - uuid; + return this + StringUtils.SPACE + StringUtils.VERTICAL_BAR + StringUtils.SPACE + uuid; } @Override @@ -145,7 +142,7 @@ public class RegisterVO { if (Objects.nonNull(providerConfig)) { var providerAddress = providerConfig.getAddress(); if (StringUtils.isBlank(providerAddress)) { - throw new RuntimeException(StringUtils.format("providerConfig的address不能为空")); + throw new RuntimeException(StringUtils.format("The address of provider Config cannot be empty")); } builder.append(StringUtils.SPACE).append(StringUtils.VERTICAL_BAR).append(StringUtils.SPACE); // 服务提供者地址 @@ -153,7 +150,7 @@ public class RegisterVO { builder.append(StringUtils.SPACE).append(StringUtils.VERTICAL_BAR).append(StringUtils.SPACE); var providerModules = providerConfig.getProviders().stream() - .map(it -> StringUtils.joinWith(StringUtils.HYPHEN, it.getProtocolModule().getId(), it.getProtocolModule().getName(), it.getProvider())) + .map(it -> StringUtils.joinWith(StringUtils.HYPHEN, it.getProtocolModule(), it.getProvider())) .toList(); // 服务提供者模块信息列表 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 ded1c9a6..23c38908 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 @@ -357,9 +357,7 @@ public class ZookeeperRegistry implements IRegistry { } /** - * 如果自己是服务提供者，就讲自己注册上去 - * - * @throws Exception + * 如果自己是服务提供者，就把自己注册上去 */ private void initLocalProvider() throws Exception { var localRegisterVO = NetContext.getConfigManager().getLocalConfig().toLocalRegisterVO(); diff --git a/net/src/test/java/com/zfoo/net/config/RegistryTest.java b/net/src/test/java/com/zfoo/net/config/RegistryTest.java index 37ed4db6..b9785206 100644 --- a/net/src/test/java/com/zfoo/net/config/RegistryTest.java +++ b/net/src/test/java/com/zfoo/net/config/RegistryTest.java @@ -34,8 +34,8 @@ public class RegistryTest { @Test public void registerVoTest() { // 定义2个模块：可以为服务提供者用，也可以为服务消费者用，这个仅仅是模块信息 - var protocolModule1 = new ProtocolModule((byte) 100, "aaa"); - var protocolModule2 = new ProtocolModule((byte) 120, "bbb"); + var protocolModule1 = "aaa"; + var protocolModule2 ="bbb"; // 服务提供者模块列表和服务提供者配置 // 定义2个服务提供者模块 @@ -44,7 +44,7 @@ public class RegistryTest { var providerConfig = ProviderConfig.valueOf(HostAndPort.valueOf("127.0.0.1", 80).toHostAndPortStr(), providerModules); // 服务消费者模块和服务消费者配置(服务消费者模块多一个负载均衡属性) - var consumerModules = List.of(new ConsumerModule(protocolModule1, "random", "a"), new ConsumerModule(protocolModule2, "random", "b")); + var consumerModules = List.of(new ConsumerModule("random", "a"), new ConsumerModule("random", "b")); // 服务消费者配置：这个是没Ip的 var consumerConfig = ConsumerConfig.valueOf(consumerModules);