diff --git a/net/src/main/java/com/zfoo/net/config/manager/ConfigManager.java b/net/src/main/java/com/zfoo/net/config/manager/ConfigManager.java index c0f7a900..a89960d0 100644 --- a/net/src/main/java/com/zfoo/net/config/manager/ConfigManager.java +++ b/net/src/main/java/com/zfoo/net/config/manager/ConfigManager.java @@ -14,8 +14,6 @@ package com.zfoo.net.config.manager; import com.zfoo.net.config.model.NetConfig; -import com.zfoo.net.consumer.balancer.AbstractConsumerLoadBalancer; -import com.zfoo.net.consumer.balancer.IConsumerLoadBalancer; import com.zfoo.net.consumer.registry.IRegistry; import com.zfoo.net.consumer.registry.ZookeeperRegistry; import com.zfoo.protocol.ProtocolManager; @@ -25,10 +23,8 @@ import com.zfoo.protocol.util.AssertionUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.ArrayList; -import java.util.Map; +import java.util.HashSet; import java.util.Objects; -import java.util.concurrent.ConcurrentHashMap; /** * @author jaysunxiao @@ -63,14 +59,30 @@ public class ConfigManager implements IConfigManager { var providerConfig = localConfig.getProvider(); if (Objects.nonNull(providerConfig) && CollectionUtils.isNotEmpty(providerConfig.getProviders())) { // 检查并且替换配置文件中的ProtocolModule - for (var provider : providerConfig.getProviders()) { - var protocolModuleName = provider.getProtocolModule().getName(); + var set = new HashSet(); + for (var providerModule : providerConfig.getProviders()) { + var provider = providerModule.getProvider(); + var protocolModuleName = providerModule.getProtocolModule().getName(); var protocolModule = ProtocolManager.moduleByModuleName(protocolModuleName); AssertionUtils.isTrue(protocolModule != null, "服务提供者[name:{}]在协议文件中不存在", protocolModuleName); - provider.setProtocolModule(protocolModule); + providerModule.setProtocolModule(protocolModule); + AssertionUtils.isTrue(set.add(provider), "服务提供者[name:{}]配置重复", provider); + } + } + var consumerConfig = localConfig.getConsumer(); + if (Objects.nonNull(consumerConfig) && CollectionUtils.isNotEmpty(consumerConfig.getConsumers())) { + var set = new HashSet(); + var protocolModuleSet = new HashSet(); + for (var consumerModule : consumerConfig.getConsumers()) { + var consumer = consumerModule.getConsumer(); + var protocolModuleName = consumerModule.getProtocolModule().getName(); + var protocolModule = ProtocolManager.moduleByModuleName(protocolModuleName); + AssertionUtils.isTrue(protocolModule != null, "服务消费者[name:{}]在协议文件中不存在", protocolModuleName); + consumerModule.setProtocolModule(protocolModule); + AssertionUtils.isTrue(set.add(consumer), "服务消费者[name:{}]配置重复", consumer); + AssertionUtils.isTrue(protocolModuleSet.add(protocolModule), "服务消费者[name:{}]重复消费了协议模块{}", consumer, protocolModule); } } - registry = new ZookeeperRegistry(); registry.start(); } @@ -79,5 +91,4 @@ public class ConfigManager implements IConfigManager { public IRegistry getRegistry() { return registry; } - } diff --git a/net/src/main/java/com/zfoo/net/config/model/ConsumerModule.java b/net/src/main/java/com/zfoo/net/config/model/ConsumerModule.java index 7b8d4883..8b75789e 100644 --- a/net/src/main/java/com/zfoo/net/config/model/ConsumerModule.java +++ b/net/src/main/java/com/zfoo/net/config/model/ConsumerModule.java @@ -14,18 +14,29 @@ package com.zfoo.net.config.model; import com.zfoo.protocol.registration.ProtocolModule; +import java.util.Objects; + /** * @author jaysunxiao * @version 3.0 */ public class ConsumerModule { - // 消费哪个provider - private String consumer; + private ProtocolModule protocolModule; private String loadBalancer; - public ConsumerModule(String consumer, String loadBalancer) { + // 消费哪个provider + private String consumer; + + public ConsumerModule(ProtocolModule protocolModule, String loadBalancer, String consumer) { + this.protocolModule = protocolModule; + this.consumer = consumer; + this.loadBalancer = loadBalancer; + } + + public ConsumerModule(String protocolModule, String loadBalancer, String consumer) { + this.protocolModule = new ProtocolModule((byte) 0, protocolModule); this.consumer = consumer; this.loadBalancer = loadBalancer; } @@ -45,4 +56,29 @@ public class ConsumerModule { public void setLoadBalancer(String loadBalancer) { this.loadBalancer = loadBalancer; } + + public ProtocolModule getProtocolModule() { + return protocolModule; + } + + public void setProtocolModule(ProtocolModule protocolModule) { + this.protocolModule = protocolModule; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + ConsumerModule that = (ConsumerModule) o; + return Objects.equals(protocolModule, that.protocolModule) && Objects.equals(consumer, that.consumer); + } + + @Override + public int hashCode() { + return Objects.hash(protocolModule, loadBalancer, consumer); + } } 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 dda51e5c..fab8dd05 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 @@ -14,24 +14,26 @@ package com.zfoo.net.config.model; import com.zfoo.protocol.registration.ProtocolModule; +import java.util.Objects; + /** * @author jaysunxiao * @version 3.0 */ public class ProviderModule { - private String provider; - private ProtocolModule protocolModule; - public ProviderModule(String provider, ProtocolModule protocolModule) { - this.provider = provider; + private String provider; + + public ProviderModule(ProtocolModule protocolModule, String provider) { this.protocolModule = protocolModule; + this.provider = provider; } - public ProviderModule(String provider, String protocolModule) { - this.provider = provider; + public ProviderModule(String protocolModule, String provider) { this.protocolModule = new ProtocolModule((byte) 0, protocolModule); + this.provider = provider; } public String getProvider() { @@ -49,4 +51,21 @@ public class ProviderModule { public void setProtocolModule(ProtocolModule protocolModule) { this.protocolModule = protocolModule; } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + ProviderModule that = (ProviderModule) o; + return Objects.equals(protocolModule, that.protocolModule) && Objects.equals(provider, that.provider); + } + + @Override + public int hashCode() { + return Objects.hash(protocolModule, provider); + } } 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 e30bf664..4cba81d3 100644 --- a/net/src/main/java/com/zfoo/net/consumer/Consumer.java +++ b/net/src/main/java/com/zfoo/net/consumer/Consumer.java @@ -58,35 +58,17 @@ public class Consumer implements IConsumer { private static final Logger logger = LoggerFactory.getLogger(Consumer.class); - private IConsumerLoadBalancer loadBalancer(ProtocolModule protocolModule) { + @Override + public IConsumerLoadBalancer loadBalancer(ProtocolModule protocolModule) { var consumerConfig = NetContext.getConfigManager().getLocalConfig().getConsumer(); if (consumerConfig == null || CollectionUtils.isEmpty(consumerConfig.getConsumers())) { throw new RunException("没有配置服务消费者，无法消费"); } var consumers = consumerConfig.getConsumers(); - var clientSessionMap = NetContext.getSessionManager().getClientSessionMap(); - for (var clientSession : clientSessionMap.values()) { - var attribute = clientSession.getAttribute(AttributeType.CONSUMER); - if (attribute == null) { - continue; - } - - var registerVO = (RegisterVO) attribute; - var providerConfig = registerVO.getProviderConfig(); - if (providerConfig == null) { - continue; - } - - for (var provider : providerConfig.getProviders()) { - if (provider.getProtocolModule().getId() != protocolModule.getId()) { - continue; - } - for (var consumer : consumers) { - if (consumer.getConsumer().equals(provider.getProvider())) { - return AbstractConsumerLoadBalancer.valueOf(consumer.getLoadBalancer()); - } - } + for (var consumer : consumers) { + if (consumer.getProtocolModule().equals(protocolModule)) { + return AbstractConsumerLoadBalancer.valueOf(consumer.getLoadBalancer()); } } return null; 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 f4653c10..d1449963 100644 --- a/net/src/main/java/com/zfoo/net/consumer/IConsumer.java +++ b/net/src/main/java/com/zfoo/net/consumer/IConsumer.java @@ -13,9 +13,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.IPacket; +import com.zfoo.protocol.registration.ProtocolModule; import org.springframework.lang.Nullable; /** @@ -24,6 +26,8 @@ import org.springframework.lang.Nullable; */ public interface IConsumer { + IConsumerLoadBalancer loadBalancer(ProtocolModule protocolModule); + /** * 直接发送，不需要任何返回值 * 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 02a7c4d9..16f1d183 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 @@ -20,7 +20,6 @@ import com.zfoo.net.config.model.ProviderModule; import com.zfoo.protocol.ProtocolManager; import com.zfoo.protocol.collection.CollectionUtils; import com.zfoo.protocol.exception.ExceptionUtils; -import com.zfoo.protocol.registration.ProtocolModule; import com.zfoo.protocol.util.StringUtils; import com.zfoo.util.security.IdUtils; import org.slf4j.Logger; @@ -47,13 +46,17 @@ public class RegisterVO { private ConsumerConfig consumerConfig; - public static boolean providerHasConsumerModule(RegisterVO provider, RegisterVO consumer) { - if (Objects.isNull(provider) || Objects.isNull(provider.providerConfig) || CollectionUtils.isEmpty(provider.providerConfig.getProviders()) - || Objects.isNull(consumer) || Objects.isNull(consumer.consumerConfig) || CollectionUtils.isEmpty(consumer.consumerConfig.getConsumers())) { + public static boolean providerHasConsumerModule(RegisterVO providerVO, RegisterVO consumerVO) { + if (Objects.isNull(providerVO) || Objects.isNull(providerVO.providerConfig) || CollectionUtils.isEmpty(providerVO.providerConfig.getProviders()) + || Objects.isNull(consumerVO) || Objects.isNull(consumerVO.consumerConfig) || CollectionUtils.isEmpty(consumerVO.consumerConfig.getConsumers())) { return false; } - - return provider.getProviderConfig().getProviders().stream().anyMatch(it -> consumer.getConsumerConfig().getConsumers().contains(it)); + for (var provider : providerVO.getProviderConfig().getProviders()) { + if (consumerVO.getConsumerConfig().getConsumers().stream().anyMatch(it -> it.getConsumer().equals(provider.getProvider()))) { + return true; + } + } + return false; } public static RegisterVO valueOf(String id, ProviderConfig providerConfig, ConsumerConfig consumerConfig) { @@ -74,7 +77,7 @@ public class RegisterVO { String providerAddress = null; - for (int i = 1; i < splits.length; i++) { + for (var i = 1; i < splits.length; i++) { var s = splits[i].trim(); if (s.startsWith("provider")) { var providerModules = parseProviderModules(s); @@ -103,7 +106,7 @@ public class RegisterVO { var modules = Arrays.stream(moduleSplits) .map(it -> it.trim()) .map(it -> it.split(StringUtils.HYPHEN)) - .map(it -> new ProviderModule(StringUtils.trim(it[1]), ProtocolManager.moduleByModuleName(StringUtils.trim(it[0])))) + .map(it -> new ProviderModule(ProtocolManager.moduleByModuleName(StringUtils.trim(it[0])), StringUtils.trim(it[1]))) .collect(Collectors.toList()); return modules; } @@ -116,7 +119,7 @@ public class RegisterVO { var modules = Arrays.stream(moduleSplits) .map(it -> it.trim()) .map(it -> it.split(StringUtils.HYPHEN)) - .map(it -> new ConsumerModule(StringUtils.trim(it[0]), StringUtils.trim(it[1]))) + .map(it -> new ConsumerModule(ProtocolManager.moduleByModuleName(StringUtils.trim(it[0])), StringUtils.trim(it[1]), StringUtils.trim(it[2]))) .collect(Collectors.toList()); return modules; } @@ -156,7 +159,7 @@ public class RegisterVO { builder.append(StringUtils.SPACE).append(StringUtils.VERTICAL_BAR).append(StringUtils.SPACE); var consumerModules = consumerConfig.getConsumers().stream() - .map(it -> StringUtils.joinWith(StringUtils.HYPHEN, it.getConsumer(), it.getLoadBalancer())) + .map(it -> StringUtils.joinWith(StringUtils.HYPHEN, it.getProtocolModule().getName(), it.getLoadBalancer(), it.getConsumer())) .collect(Collectors.toList()); builder.append(StringUtils.format("consumer:[{}]" , StringUtils.joinWith(StringUtils.COMMA + StringUtils.SPACE, consumerModules.toArray()))); 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 dc5d1650..456f8eda 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 @@ -391,7 +391,7 @@ public class ZookeeperRegistry implements IRegistry { return; } - logger.info("开始通过[providerCacheSet:{}]检查[consumer:{}]", providerCacheSet, NetContext.getSessionManager().getClientSessionMap().size()); + logger.info("开始通过providerCacheSet:{}检查[consumer:{}]", providerCacheSet, NetContext.getSessionManager().getClientSessionMap().size()); var recheckFlag = false; diff --git a/net/src/main/java/com/zfoo/net/schema/NetDefinitionParser.java b/net/src/main/java/com/zfoo/net/schema/NetDefinitionParser.java index 4bd31362..2647f607 100644 --- a/net/src/main/java/com/zfoo/net/schema/NetDefinitionParser.java +++ b/net/src/main/java/com/zfoo/net/schema/NetDefinitionParser.java @@ -182,8 +182,8 @@ public class NetDefinitionParser implements BeanDefinitionParser { var clazz = ProviderModule.class; var builder = BeanDefinitionBuilder.rootBeanDefinition(clazz); - builder.addConstructorArgValue(environment.resolvePlaceholders(addressElement.getAttribute("provider"))); builder.addConstructorArgValue(environment.resolvePlaceholders(addressElement.getAttribute("protocol-module"))); + builder.addConstructorArgValue(environment.resolvePlaceholders(addressElement.getAttribute("provider"))); providers.add(new BeanDefinitionHolder(builder.getBeanDefinition(), StringUtils.format("{}.{}{}", clazz.getCanonicalName(), param, i))); } @@ -199,8 +199,9 @@ public class NetDefinitionParser implements BeanDefinitionParser { var clazz = ConsumerModule.class; var builder = BeanDefinitionBuilder.rootBeanDefinition(clazz); - builder.addConstructorArgValue(environment.resolvePlaceholders(addressElement.getAttribute("consumer"))); + builder.addConstructorArgValue(environment.resolvePlaceholders(addressElement.getAttribute("protocol-module"))); builder.addConstructorArgValue(environment.resolvePlaceholders(addressElement.getAttribute("load-balancer"))); + builder.addConstructorArgValue(environment.resolvePlaceholders(addressElement.getAttribute("consumer"))); modules.add(new BeanDefinitionHolder(builder.getBeanDefinition(), StringUtils.format("{}.{}{}", clazz.getCanonicalName(), param, i))); } diff --git a/net/src/main/resources/net-1.0.xsd b/net/src/main/resources/net-1.0.xsd index b5163250..2c6ba935 100644 --- a/net/src/main/resources/net-1.0.xsd +++ b/net/src/main/resources/net-1.0.xsd @@ -44,8 +44,9 @@ - + + diff --git a/net/src/test/java/com/zfoo/net/core/provider/ProviderTest.java b/net/src/test/java/com/zfoo/net/core/provider/ProviderTest.java index eb40145c..1a5b6113 100644 --- a/net/src/test/java/com/zfoo/net/core/provider/ProviderTest.java +++ b/net/src/test/java/com/zfoo/net/core/provider/ProviderTest.java @@ -149,25 +149,4 @@ public class ProviderTest { ThreadUtils.sleep(Long.MAX_VALUE); } - /** - * 固定消费方式 - */ - @Test - public void startFixedConsumer() { - var context = new ClassPathXmlApplicationContext("provider/consumer_fixed_config.xml"); - SessionUtils.printSessionInfo(); - - var ask = new ProviderMessAsk(); - ask.setMessage("Hello, this is the consumer!"); - var atomicInteger = new AtomicInteger(0); - - for (int i = 0; i < 1000; i++) { - ThreadUtils.sleep(3000); - NetContext.getConsumer().asyncAsk(ask, ProviderMessAnswer.class, 0).whenComplete(answer -> { - logger.info("消费者请求[{}]收到消息[{}]", atomicInteger.incrementAndGet(), JsonUtils.object2String(answer)); - }); - } - - ThreadUtils.sleep(Long.MAX_VALUE); - } } diff --git a/net/src/test/resources/provider/consumer_consistent_session_config.xml b/net/src/test/resources/provider/consumer_consistent_session_config.xml index b73b9222..056fac7f 100644 --- a/net/src/test/resources/provider/consumer_consistent_session_config.xml +++ b/net/src/test/resources/provider/consumer_consistent_session_config.xml @@ -20,13 +20,12 @@ - - + diff --git a/net/src/test/resources/provider/consumer_random_config.xml b/net/src/test/resources/provider/consumer_random_config.xml index b4cd397d..6c6f6883 100644 --- a/net/src/test/resources/provider/consumer_random_config.xml +++ b/net/src/test/resources/provider/consumer_random_config.xml @@ -19,13 +19,12 @@ - - + diff --git a/net/src/test/resources/provider/consumer_shortest_time_config.xml b/net/src/test/resources/provider/consumer_shortest_time_config.xml index aa032106..6894c171 100644 --- a/net/src/test/resources/provider/consumer_shortest_time_config.xml +++ b/net/src/test/resources/provider/consumer_shortest_time_config.xml @@ -19,16 +19,13 @@ - - - + - diff --git a/net/src/test/resources/provider/provider_config.xml b/net/src/test/resources/provider/provider_config.xml index 3c69a79a..e54749f9 100644 --- a/net/src/test/resources/provider/provider_config.xml +++ b/net/src/test/resources/provider/provider_config.xml @@ -20,15 +20,14 @@ - - + + -