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 44070411..c0f7a900 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 @@ -18,7 +18,6 @@ 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.net.session.model.Session; import com.zfoo.protocol.ProtocolManager; import com.zfoo.protocol.collection.CollectionUtils; import com.zfoo.protocol.registration.ProtocolModule; @@ -27,7 +26,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.ArrayList; -import java.util.HashMap; import java.util.Map; import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; @@ -45,10 +43,6 @@ public class ConfigManager implements IConfigManager { */ private NetConfig localConfig; - private AbstractConsumerLoadBalancer consumerLoadBalancer; - - private final Map consumerLoadBalancerMap = new ConcurrentHashMap<>(); - /** * 注册中心 */ @@ -63,40 +57,18 @@ public class ConfigManager implements IConfigManager { this.localConfig = localConfig; } - @Override - public IConsumerLoadBalancer consumerLoadBalancer(ProtocolModule module) { - return consumerLoadBalancerMap.get(module.getName()); - } - @Override public void initRegistry() { // 通过protocol，写入provider的module的id和version var providerConfig = localConfig.getProvider(); - if (Objects.nonNull(providerConfig) && CollectionUtils.isNotEmpty(providerConfig.getModules())) { - var providerModules = new ArrayList(providerConfig.getModules().size()); - for (var providerModule : providerConfig.getModules()) { - var module = ProtocolManager.moduleByModuleName(providerModule.getName()); - AssertionUtils.isTrue(module != null, "服务提供者[name:{}]在协议文件中不存在", providerModule.getName()); - module.setGroup(providerModule.getGroup()); - providerModules.add(module); + if (Objects.nonNull(providerConfig) && CollectionUtils.isNotEmpty(providerConfig.getProviders())) { + // 检查并且替换配置文件中的ProtocolModule + for (var provider : providerConfig.getProviders()) { + var protocolModuleName = provider.getProtocolModule().getName(); + var protocolModule = ProtocolManager.moduleByModuleName(protocolModuleName); + AssertionUtils.isTrue(protocolModule != null, "服务提供者[name:{}]在协议文件中不存在", protocolModuleName); + provider.setProtocolModule(protocolModule); } - providerConfig.setModules(providerModules); - } - - // 通过protocol，写入consumer的module的id和version - var consumerConfig = localConfig.getConsumer(); - if (Objects.nonNull(consumerConfig) && CollectionUtils.isNotEmpty(consumerConfig.getModules())) { - var consumerModules = new ArrayList(consumerConfig.getModules().size()); - for (var providerModule : consumerConfig.getModules()) { - var module = ProtocolManager.moduleByModuleName(providerModule.getName()); - AssertionUtils.isTrue(module != null, "消费者[name:{}]在协议文件中不存在", providerModule.getName()); - module.setGroup(providerModule.getGroup()); - module.setLoadBalancer(providerModule.getLoadBalancer()); - consumerModules.add(module); - consumerLoadBalancerMap.put(module.getName(), AbstractConsumerLoadBalancer.valueOf(module.getLoadBalancer())); - } - consumerConfig.setModules(consumerModules); -// consumerLoadBalancer = AbstractConsumerLoadBalancer.valueOf(consumerConfig.getLoadBalancer()); } registry = new ZookeeperRegistry(); diff --git a/net/src/main/java/com/zfoo/net/config/manager/IConfigManager.java b/net/src/main/java/com/zfoo/net/config/manager/IConfigManager.java index 256c168e..4008df00 100644 --- a/net/src/main/java/com/zfoo/net/config/manager/IConfigManager.java +++ b/net/src/main/java/com/zfoo/net/config/manager/IConfigManager.java @@ -27,8 +27,6 @@ public interface IConfigManager { NetConfig getLocalConfig(); - IConsumerLoadBalancer consumerLoadBalancer(ProtocolModule module); - void initRegistry(); IRegistry getRegistry(); diff --git a/net/src/main/java/com/zfoo/net/config/model/ConsumerConfig.java b/net/src/main/java/com/zfoo/net/config/model/ConsumerConfig.java index a5dfd33a..003fc7c4 100644 --- a/net/src/main/java/com/zfoo/net/config/model/ConsumerConfig.java +++ b/net/src/main/java/com/zfoo/net/config/model/ConsumerConfig.java @@ -25,20 +25,20 @@ import java.util.Objects; */ public class ConsumerConfig { - private List modules; + private List consumers; - public static ConsumerConfig valueOf(List modules) { + public static ConsumerConfig valueOf(List modules) { ConsumerConfig config = new ConsumerConfig(); - config.modules = modules; + config.consumers = modules; return config; } - public List getModules() { - return modules; + public List getConsumers() { + return consumers; } - public void setModules(List modules) { - this.modules = modules; + public void setConsumers(List consumers) { + this.consumers = consumers; } @Override @@ -50,11 +50,11 @@ public class ConsumerConfig { return false; } ConsumerConfig that = (ConsumerConfig) o; - return Objects.equals(modules, that.modules); + return Objects.equals(consumers, that.consumers); } @Override public int hashCode() { - return Objects.hash(modules); + return Objects.hash(consumers); } } 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 new file mode 100644 index 00000000..7b8d4883 --- /dev/null +++ b/net/src/main/java/com/zfoo/net/config/model/ConsumerModule.java @@ -0,0 +1,48 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.net.config.model; + +import com.zfoo.protocol.registration.ProtocolModule; + +/** + * @author jaysunxiao + * @version 3.0 + */ +public class ConsumerModule { + + // 消费哪个provider + private String consumer; + + private String loadBalancer; + + public ConsumerModule(String consumer, String loadBalancer) { + this.consumer = consumer; + this.loadBalancer = loadBalancer; + } + + public String getConsumer() { + return consumer; + } + + public void setConsumer(String consumer) { + this.consumer = consumer; + } + + public String getLoadBalancer() { + return loadBalancer; + } + + public void setLoadBalancer(String loadBalancer) { + this.loadBalancer = loadBalancer; + } +} 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 e01ecf6b..757a086d 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 @@ -13,7 +13,6 @@ package com.zfoo.net.config.model; -import com.zfoo.protocol.registration.ProtocolModule; import com.zfoo.protocol.util.StringUtils; import com.zfoo.util.net.HostAndPort; import com.zfoo.util.net.NetUtils; @@ -38,12 +37,12 @@ public class ProviderConfig { private String address; - private List modules; + private List providers; - public static ProviderConfig valueOf(String address, List modules) { + public static ProviderConfig valueOf(String address, List modules) { ProviderConfig config = new ProviderConfig(); config.address = address; - config.modules = modules; + config.providers = modules; return config; } @@ -80,12 +79,12 @@ public class ProviderConfig { this.address = address; } - public List getModules() { - return modules; + public List getProviders() { + return providers; } - public void setModules(List modules) { - this.modules = modules; + public void setProviders(List providers) { + this.providers = providers; } @Override @@ -97,11 +96,11 @@ public class ProviderConfig { return false; } ProviderConfig that = (ProviderConfig) o; - return Objects.equals(address, that.address) && Objects.equals(modules, that.modules); + return Objects.equals(address, that.address) && Objects.equals(providers, that.providers); } @Override public int hashCode() { - return Objects.hash(address, modules); + return Objects.hash(address, 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 new file mode 100644 index 00000000..dda51e5c --- /dev/null +++ b/net/src/main/java/com/zfoo/net/config/model/ProviderModule.java @@ -0,0 +1,52 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.net.config.model; + +import com.zfoo.protocol.registration.ProtocolModule; + +/** + * @author jaysunxiao + * @version 3.0 + */ +public class ProviderModule { + + private String provider; + + private ProtocolModule protocolModule; + + public ProviderModule(String provider, ProtocolModule protocolModule) { + this.provider = provider; + this.protocolModule = protocolModule; + } + + public ProviderModule(String provider, String protocolModule) { + this.provider = provider; + this.protocolModule = new ProtocolModule((byte) 0, protocolModule); + } + + public String getProvider() { + return provider; + } + + public void setProvider(String provider) { + this.provider = provider; + } + + public ProtocolModule getProtocolModule() { + return protocolModule; + } + + public void setProtocolModule(ProtocolModule protocolModule) { + this.protocolModule = protocolModule; + } +} 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 68f9c0a5..e30bf664 100644 --- a/net/src/main/java/com/zfoo/net/consumer/Consumer.java +++ b/net/src/main/java/com/zfoo/net/consumer/Consumer.java @@ -14,6 +14,9 @@ package com.zfoo.net.consumer; import com.zfoo.net.NetContext; +import com.zfoo.net.consumer.balancer.AbstractConsumerLoadBalancer; +import com.zfoo.net.consumer.balancer.IConsumerLoadBalancer; +import com.zfoo.net.consumer.registry.RegisterVO; import com.zfoo.net.packet.common.Error; import com.zfoo.net.router.Router; import com.zfoo.net.router.answer.AsyncAnswer; @@ -24,8 +27,13 @@ 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.route.SignalBridge; +import com.zfoo.net.session.model.AttributeType; +import com.zfoo.net.session.model.Session; import com.zfoo.protocol.IPacket; import com.zfoo.protocol.ProtocolManager; +import com.zfoo.protocol.collection.CollectionUtils; +import com.zfoo.protocol.exception.RunException; +import com.zfoo.protocol.registration.ProtocolModule; import com.zfoo.protocol.util.JsonUtils; import com.zfoo.protocol.util.StringUtils; import com.zfoo.util.math.HashUtils; @@ -33,6 +41,8 @@ import com.zfoo.util.math.RandomUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -48,12 +58,45 @@ public class Consumer implements IConsumer { private static final Logger logger = LoggerFactory.getLogger(Consumer.class); + private 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()); + } + } + } + } + return null; + } @Override public void send(IPacket packet, Object argument) { try { var module = ProtocolManager.moduleByProtocolId(packet.protocolId()); - var loadBalancer = NetContext.getConfigManager().consumerLoadBalancer(module); + var loadBalancer = loadBalancer(module); var session = loadBalancer.loadBalancer(packet, argument); var executorConsistentHash = (argument == null) ? RandomUtils.randomInt() : HashUtils.fnvHash(argument); NetContext.getRouter().send(session, packet, NoAnswerAttachment.valueOf(executorConsistentHash)); @@ -65,7 +108,7 @@ public class Consumer implements IConsumer { @Override public SyncAnswer syncAsk(IPacket packet, Class answerClass, Object argument) throws Exception { var module = ProtocolManager.moduleByProtocolId(packet.protocolId()); - var loadBalancer = NetContext.getConfigManager().consumerLoadBalancer(module); + var loadBalancer = loadBalancer(module); var session = loadBalancer.loadBalancer(packet, argument); @@ -107,7 +150,7 @@ public class Consumer implements IConsumer { @Override public AsyncAnswer asyncAsk(IPacket packet, Class answerClass, Object argument) { var module = ProtocolManager.moduleByProtocolId(packet.protocolId()); - var loadBalancer = NetContext.getConfigManager().consumerLoadBalancer(module); + var loadBalancer = loadBalancer(module); var session = loadBalancer.loadBalancer(packet, argument); var asyncAnswer = NetContext.getRouter().asyncAsk(session, packet, answerClass, argument); 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 aa1096f7..6c0319bf 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 @@ -19,9 +19,11 @@ import com.zfoo.net.session.model.AttributeType; import com.zfoo.net.session.model.Session; import com.zfoo.protocol.IPacket; import com.zfoo.protocol.ProtocolManager; +import com.zfoo.protocol.model.Pair; 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.stream.Collectors; @@ -44,9 +46,6 @@ public abstract class AbstractConsumerLoadBalancer implements IConsumerLoadBalan case "shortest-time": balancer = ShortestTimeConsumerLoadBalancer.getInstance(); break; - case "fixed": - balancer = FixedConsumerLoadBalancer.getInstance(); - break; default: throw new RuntimeException(StringUtils.format("无法识别负载均衡器[{}]", loadBalancer)); } @@ -64,11 +63,7 @@ public abstract class AbstractConsumerLoadBalancer implements IConsumerLoadBalan var attribute = it.getAttribute(AttributeType.CONSUMER); if (Objects.nonNull(attribute)) { var registerVO = (RegisterVO) attribute; - if (Objects.nonNull(registerVO.getProviderConfig()) && registerVO.getProviderConfig().getModules().contains(module)) { - return true; - } else { - return false; - } + return Objects.nonNull(registerVO.getProviderConfig()) && registerVO.getProviderConfig().getProviders().stream().anyMatch(provider -> provider.getProtocolModule().getId() == module.getId()); } else { return false; } @@ -77,6 +72,28 @@ public abstract class AbstractConsumerLoadBalancer implements IConsumerLoadBalan return sessions; } + public List sessionsByModule(ProtocolModule module) { + var clientSessionMap = NetContext.getSessionManager().getClientSessionMap(); + var sessions = new ArrayList(); + 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; + } + + if (providerConfig.getProviders().stream().anyMatch(it -> it.getProtocolModule().getId() == module.getId())) { + sessions.add(clientSession); + } + } + return sessions; + } + public boolean sessionHasModule(Session session, IPacket packet) { @@ -91,6 +108,6 @@ public abstract class AbstractConsumerLoadBalancer implements IConsumerLoadBalan } var module = ProtocolManager.moduleByProtocolId(packet.protocolId()); - return registerVO.getProviderConfig().getModules().contains(module); + return registerVO.getProviderConfig().getProviders().contains(module); } } diff --git a/net/src/main/java/com/zfoo/net/consumer/balancer/FixedConsumerLoadBalancer.java b/net/src/main/java/com/zfoo/net/consumer/balancer/FixedConsumerLoadBalancer.java deleted file mode 100644 index 32992970..00000000 --- a/net/src/main/java/com/zfoo/net/consumer/balancer/FixedConsumerLoadBalancer.java +++ /dev/null @@ -1,61 +0,0 @@ -/* - * Copyright (C) 2020 The zfoo Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except - * in compliance with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed - * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and limitations under the License. - */ - -package com.zfoo.net.consumer.balancer; - -import com.zfoo.net.consumer.registry.RegisterVO; -import com.zfoo.net.session.model.AttributeType; -import com.zfoo.net.session.model.Session; -import com.zfoo.protocol.IPacket; -import com.zfoo.protocol.ProtocolManager; -import com.zfoo.protocol.exception.RunException; - -/** - * 根据grouId获取固定服务器 - * - * @author jaysunxiao - * @version 3.0 - */ -public class FixedConsumerLoadBalancer extends AbstractConsumerLoadBalancer { - - private static final FixedConsumerLoadBalancer INSTANCE = new FixedConsumerLoadBalancer(); - - private FixedConsumerLoadBalancer() { - } - - public static FixedConsumerLoadBalancer getInstance() { - return INSTANCE; - } - - @Override - public Session loadBalancer(IPacket packet, Object argument) { - var module = ProtocolManager.moduleByProtocolId(packet.protocolId()); - var sessions = getSessionsByModule(module); - - if (sessions.isEmpty()) { - throw new RunException("获取固定服务器失败[protocolId:{}]参数[argument:{}],没有服务提供者提供服务[module:{}]", packet.protocolId(), argument, module); - } - - int group = Integer.valueOf(argument.toString()); - for (var session : sessions) { - var registerVO = (RegisterVO)session.getAttribute(AttributeType.CONSUMER); - var isPresent = registerVO.getProviderConfig().getModules().stream().filter(it -> it.getName().equals(module.getName()) && it.getGroup() == group).findAny().isPresent(); - if (!isPresent) { - continue; - } - return session; - } - throw new RunException("一获取固定服务器失败[protocolId:{}]参数[argument:{}],没有服务提供者提供服务[module:{}]", packet.protocolId(), argument, module); - } - -} 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 c6e28027..02a7c4d9 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 @@ -14,7 +14,10 @@ package com.zfoo.net.consumer.registry; import com.zfoo.net.config.model.ConsumerConfig; +import com.zfoo.net.config.model.ConsumerModule; import com.zfoo.net.config.model.ProviderConfig; +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; @@ -45,12 +48,12 @@ public class RegisterVO { public static boolean providerHasConsumerModule(RegisterVO provider, RegisterVO consumer) { - if (Objects.isNull(provider) || Objects.isNull(provider.providerConfig) || CollectionUtils.isEmpty(provider.providerConfig.getModules()) - || Objects.isNull(consumer) || Objects.isNull(consumer.consumerConfig) || CollectionUtils.isEmpty(consumer.consumerConfig.getModules())) { + 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())) { return false; } - return provider.getProviderConfig().getModules().stream().anyMatch(it -> consumer.getConsumerConfig().getModules().contains(it)); + return provider.getProviderConfig().getProviders().stream().anyMatch(it -> consumer.getConsumerConfig().getConsumers().contains(it)); } public static RegisterVO valueOf(String id, ProviderConfig providerConfig, ConsumerConfig consumerConfig) { @@ -74,10 +77,10 @@ public class RegisterVO { for (int i = 1; i < splits.length; i++) { var s = splits[i].trim(); if (s.startsWith("provider")) { - var providerModules = parseModules(s); + var providerModules = parseProviderModules(s); vo.providerConfig = ProviderConfig.valueOf(providerAddress, providerModules); } else if (s.startsWith("consumer")) { - var consumerModules = parseModules(s); + var consumerModules = parseConsumerModules(s); vo.consumerConfig = ConsumerConfig.valueOf(consumerModules); } else { providerAddress = s; @@ -91,7 +94,8 @@ public class RegisterVO { } } - private static List parseModules(String str) { + + private static List parseProviderModules(String str) { var moduleSplits = StringUtils.substringBeforeLast( StringUtils.substringAfterFirst(str, StringUtils.LEFT_SQUARE_BRACKET) , StringUtils.RIGHT_SQUARE_BRACKET).split(StringUtils.COMMA); @@ -99,7 +103,20 @@ public class RegisterVO { var modules = Arrays.stream(moduleSplits) .map(it -> it.trim()) .map(it -> it.split(StringUtils.HYPHEN)) - .map(it -> new ProtocolModule(Byte.parseByte(it[0]), it[1], Integer.parseInt(it[2]))) + .map(it -> new ProviderModule(StringUtils.trim(it[1]), ProtocolManager.moduleByModuleName(StringUtils.trim(it[0])))) + .collect(Collectors.toList()); + return modules; + } + + private static List parseConsumerModules(String str) { + var moduleSplits = StringUtils.substringBeforeLast( + StringUtils.substringAfterFirst(str, StringUtils.LEFT_SQUARE_BRACKET) + , StringUtils.RIGHT_SQUARE_BRACKET).split(StringUtils.COMMA); + + 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]))) .collect(Collectors.toList()); return modules; } @@ -128,8 +145,8 @@ public class RegisterVO { builder.append(providerAddress); builder.append(StringUtils.SPACE).append(StringUtils.VERTICAL_BAR).append(StringUtils.SPACE); - var providerModules = providerConfig.getModules().stream() - .map(it -> joinWith(StringUtils.HYPHEN, it)) + var providerModules = providerConfig.getProviders().stream() + .map(it -> StringUtils.joinWith(StringUtils.HYPHEN, it.getProtocolModule().getName(), it.getProvider())) .collect(Collectors.toList()); builder.append(StringUtils.format("provider:[{}]" , StringUtils.joinWith(StringUtils.COMMA + StringUtils.SPACE, providerModules.toArray()))); @@ -138,8 +155,8 @@ public class RegisterVO { if (Objects.nonNull(consumerConfig)) { builder.append(StringUtils.SPACE).append(StringUtils.VERTICAL_BAR).append(StringUtils.SPACE); - var consumerModules = consumerConfig.getModules().stream() - .map(it -> joinWith(StringUtils.HYPHEN, it)) + var consumerModules = consumerConfig.getConsumers().stream() + .map(it -> StringUtils.joinWith(StringUtils.HYPHEN, it.getConsumer(), it.getLoadBalancer())) .collect(Collectors.toList()); builder.append(StringUtils.format("consumer:[{}]" , StringUtils.joinWith(StringUtils.COMMA + StringUtils.SPACE, consumerModules.toArray()))); @@ -148,9 +165,6 @@ public class RegisterVO { return builder.toString(); } - public String joinWith(String sep, ProtocolModule module) { - return StringUtils.joinWith(sep, module.getId(), module.getName(), module.getGroup()); - } public String getId() { return id; 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 41d89649..4bd31362 100644 --- a/net/src/main/java/com/zfoo/net/schema/NetDefinitionParser.java +++ b/net/src/main/java/com/zfoo/net/schema/NetDefinitionParser.java @@ -20,7 +20,6 @@ import com.zfoo.net.consumer.Consumer; import com.zfoo.net.packet.service.PacketService; import com.zfoo.net.router.Router; import com.zfoo.net.session.manager.SessionManager; -import com.zfoo.protocol.registration.ProtocolModule; import com.zfoo.protocol.util.DomUtils; import com.zfoo.protocol.util.StringUtils; import org.springframework.beans.factory.config.BeanDefinitionHolder; @@ -112,13 +111,13 @@ public class NetDefinitionParser implements BeanDefinitionParser { builder.addPropertyReference("monitor", MonitorConfig.class.getCanonicalName()); } - var providerElement = DomUtils.getFirstChildElementByTagName(element, "provider"); + var providerElement = DomUtils.getFirstChildElementByTagName(element, "providers"); if (providerElement != null) { builder.addPropertyReference("provider", ProviderConfig.class.getCanonicalName()); parseProviderConfig(providerElement, parserContext); } - var consumerElement = DomUtils.getFirstChildElementByTagName(element, "consumer"); + var consumerElement = DomUtils.getFirstChildElementByTagName(element, "consumers"); if (consumerElement != null) { parseConsumerConfig(consumerElement, parserContext); builder.addPropertyReference("consumer", ConsumerConfig.class.getCanonicalName()); @@ -159,8 +158,8 @@ public class NetDefinitionParser implements BeanDefinitionParser { resolvePlaceholder("thread", "thread", builder, element, parserContext); resolvePlaceholder("address", "address", builder, element, parserContext); - var providerModules = parseModules("provider", element, parserContext); - builder.addPropertyValue("modules", providerModules); + var providerModules = parseProviderModules("providers", element, parserContext); + builder.addPropertyValue("providers", providerModules); parserContext.getRegistry().registerBeanDefinition(clazz.getCanonicalName(), builder.getBeanDefinition()); } @@ -168,23 +167,40 @@ public class NetDefinitionParser implements BeanDefinitionParser { var clazz = ConsumerConfig.class; var builder = BeanDefinitionBuilder.rootBeanDefinition(clazz); - var consumerModules = parseModules("consumer", element, parserContext); - builder.addPropertyValue("modules", consumerModules); + var consumerModules = parseConsumerModules("consumers", element, parserContext); + builder.addPropertyValue("consumers", consumerModules); parserContext.getRegistry().registerBeanDefinition(clazz.getCanonicalName(), builder.getBeanDefinition()); } - private ManagedList parseModules(String param, Element element, ParserContext parserContext) { - var moduleElementList = DomUtils.getChildElementsByTagName(element, "module"); + + private ManagedList parseProviderModules(String param, Element element, ParserContext parserContext) { + var moduleElementList = DomUtils.getChildElementsByTagName(element, "provider"); + var providers = new ManagedList(); + var environment = parserContext.getReaderContext().getEnvironment(); + for (var i = 0; i < moduleElementList.size(); i++) { + var addressElement = moduleElementList.get(i); + var clazz = ProviderModule.class; + var builder = BeanDefinitionBuilder.rootBeanDefinition(clazz); + + builder.addConstructorArgValue(environment.resolvePlaceholders(addressElement.getAttribute("provider"))); + builder.addConstructorArgValue(environment.resolvePlaceholders(addressElement.getAttribute("protocol-module"))); + + providers.add(new BeanDefinitionHolder(builder.getBeanDefinition(), StringUtils.format("{}.{}{}", clazz.getCanonicalName(), param, i))); + } + return providers; + } + + private ManagedList parseConsumerModules(String param, Element element, ParserContext parserContext) { + var moduleElementList = DomUtils.getChildElementsByTagName(element, "consumer"); var modules = new ManagedList(); var environment = parserContext.getReaderContext().getEnvironment(); for (var i = 0; i < moduleElementList.size(); i++) { var addressElement = moduleElementList.get(i); - var clazz = ProtocolModule.class; + var clazz = ConsumerModule.class; var builder = BeanDefinitionBuilder.rootBeanDefinition(clazz); - builder.addConstructorArgValue(environment.resolvePlaceholders(addressElement.getAttribute("name"))); + builder.addConstructorArgValue(environment.resolvePlaceholders(addressElement.getAttribute("consumer"))); builder.addConstructorArgValue(environment.resolvePlaceholders(addressElement.getAttribute("load-balancer"))); - builder.addConstructorArgValue(environment.resolvePlaceholders(addressElement.getAttribute("group"))); 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 857c9ca9..b5163250 100644 --- a/net/src/main/resources/net-1.0.xsd +++ b/net/src/main/resources/net-1.0.xsd @@ -23,32 +23,30 @@ - - - - - - + - + - + - + - - - - + + + + + + + @@ -59,10 +57,10 @@ - + - + diff --git a/net/src/test/java/com/zfoo/net/config/RegistryTest.java b/net/src/test/java/com/zfoo/net/config/RegistryTest.java deleted file mode 100644 index ae55e376..00000000 --- a/net/src/test/java/com/zfoo/net/config/RegistryTest.java +++ /dev/null @@ -1,54 +0,0 @@ -/* - * Copyright (C) 2020 The zfoo Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except - * in compliance with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed - * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and limitations under the License. - */ - -package com.zfoo.net.config; - -import com.zfoo.net.config.model.ConsumerConfig; -import com.zfoo.net.config.model.ProviderConfig; -import com.zfoo.net.consumer.registry.RegisterVO; -import com.zfoo.protocol.registration.ProtocolModule; -import com.zfoo.util.net.HostAndPort; -import io.netty.util.NetUtil; -import org.junit.Assert; -import org.junit.Test; - -import java.util.List; - -/** - * @author jaysunxiao - * @version 3.0 - */ -public class RegistryTest { - - - @Test - public void registerVoTest() { - var modules = List.of(new ProtocolModule((byte) 100, "aaa") - , new ProtocolModule((byte) 120, "bbb")); - var providerConfig = ProviderConfig.valueOf(HostAndPort.valueOf("127.0.0.1", 80).toHostAndPortStr(), modules); - var consumerConfig = ConsumerConfig.valueOf(modules); - - var vo = RegisterVO.valueOf("test", providerConfig, consumerConfig); - var voStr = vo.toString(); - System.out.println(voStr); - var newVo = RegisterVO.parseString(voStr); - Assert.assertEquals(vo, newVo); - - System.out.println(NetUtil.LOCALHOST); - System.out.println(NetUtil.LOCALHOST4); - System.out.println(NetUtil.LOCALHOST6); - System.out.println(NetUtil.SOMAXCONN); - System.out.println(NetUtil.LOOPBACK_IF); - } - -} 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 55c7cdfa..b73b9222 100644 --- a/net/src/test/resources/provider/consumer_consistent_session_config.xml +++ b/net/src/test/resources/provider/consumer_consistent_session_config.xml @@ -25,10 +25,9 @@ - - - - + + + diff --git a/net/src/test/resources/provider/consumer_fixed_config.xml b/net/src/test/resources/provider/consumer_fixed_config.xml deleted file mode 100644 index 4395ad7e..00000000 --- a/net/src/test/resources/provider/consumer_fixed_config.xml +++ /dev/null @@ -1,34 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - diff --git a/net/src/test/resources/provider/consumer_random_config.xml b/net/src/test/resources/provider/consumer_random_config.xml index 956f5d47..b4cd397d 100644 --- a/net/src/test/resources/provider/consumer_random_config.xml +++ b/net/src/test/resources/provider/consumer_random_config.xml @@ -24,10 +24,9 @@ - - - - + + + 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 637355a5..aa032106 100644 --- a/net/src/test/resources/provider/consumer_shortest_time_config.xml +++ b/net/src/test/resources/provider/consumer_shortest_time_config.xml @@ -25,9 +25,9 @@ - - - + + + diff --git a/net/src/test/resources/provider/provider_config.xml b/net/src/test/resources/provider/provider_config.xml index aa248d02..3c69a79a 100644 --- a/net/src/test/resources/provider/provider_config.xml +++ b/net/src/test/resources/provider/provider_config.xml @@ -25,9 +25,9 @@ - - - + + + diff --git a/protocol/src/main/java/com/zfoo/protocol/registration/ProtocolModule.java b/protocol/src/main/java/com/zfoo/protocol/registration/ProtocolModule.java index 4a26bf9f..2d468d96 100644 --- a/protocol/src/main/java/com/zfoo/protocol/registration/ProtocolModule.java +++ b/protocol/src/main/java/com/zfoo/protocol/registration/ProtocolModule.java @@ -27,11 +27,6 @@ public class ProtocolModule { private String name; - private String loadBalancer; - - private int group; - - public ProtocolModule(byte id, String name) { if (id < 0) { throw new IllegalArgumentException(StringUtils.format("模块[{}]的id[{}]必须大于0", name, id)); @@ -41,18 +36,6 @@ public class ProtocolModule { this.name = name; } - public ProtocolModule(byte id, String name, int group) { - this.id = id; - this.name = name; - this.group = group; - } - - public ProtocolModule(String name, String loadBalancer, String group) { - this.name = name; - this.loadBalancer = loadBalancer; - this.group = Integer.parseInt(group); - } - public byte getId() { return id; @@ -70,22 +53,6 @@ public class ProtocolModule { this.name = name; } - public String getLoadBalancer() { - return loadBalancer; - } - - public void setLoadBalancer(String loadBalancer) { - this.loadBalancer = loadBalancer; - } - - public int getGroup() { - return group; - } - - public void setGroup(int group) { - this.group = group; - } - @Override public boolean equals(Object o) { if (this == o) { @@ -95,10 +62,7 @@ public class ProtocolModule { return false; } ProtocolModule module = (ProtocolModule) o; - if (group == 0 || module.group == 0) { - return id == module.id; - } - return id == module.id && group == module.group; + return id == module.id; } @Override @@ -108,6 +72,6 @@ public class ProtocolModule { @Override public String toString() { - return StringUtils.format("[id:{}][name:{}][group:{}]", id, name, group); + return StringUtils.format("[id:{}][name:{}]", id, name); } }