tiennm99/zfoo
1
0
Fork 0
mirror of https://github.com/tiennm99/zfoo.git synced 2026-09-08 06:17:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf[net]: 重构了net

Browse Source
This commit is contained in:
jaysunxiao
2022-07-01 18:38:48 +08:00
parent 1c94c2f311
commit 06155340f8
19 changed files with 280 additions and 310 deletions
Download Patch File Download Diff File Expand all files Collapse all files
net/src/main/java/com/zfoo/net/config/manager/ConfigManager.java
+7 -35
View File
@@ -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<String, IConsumerLoadBalancer> 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<ProtocolModule>(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<ProtocolModule>(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();
net/src/main/java/com/zfoo/net/config/manager/IConfigManager.java
-2
View File
@@ -27,8 +27,6 @@ public interface IConfigManager {
NetConfig getLocalConfig();
IConsumerLoadBalancer consumerLoadBalancer(ProtocolModule module);
void initRegistry();
IRegistry getRegistry();
net/src/main/java/com/zfoo/net/config/model/ConsumerConfig.java
+9 -9
View File
@@ -25,20 +25,20 @@ import java.util.Objects;
*/
public class ConsumerConfig {
private List<ProtocolModule> modules;
private List<ConsumerModule> consumers;
public static ConsumerConfig valueOf(List<ProtocolModule> modules) {
public static ConsumerConfig valueOf(List<ConsumerModule> modules) {
ConsumerConfig config = new ConsumerConfig();
config.modules = modules;
config.consumers = modules;
return config;
}
public List<ProtocolModule> getModules() {
return modules;
public List<ConsumerModule> getConsumers() {
return consumers;
}
public void setModules(List<ProtocolModule> modules) {
this.modules = modules;
public void setConsumers(List<ConsumerModule> 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);
}
}
net/src/main/java/com/zfoo/net/config/model/ConsumerModule.java
+48
View File
@@ -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;
}
}
net/src/main/java/com/zfoo/net/config/model/ProviderConfig.java
+9 -10
View File
@@ -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<ProtocolModule> modules;
private List<ProviderModule> providers;
public static ProviderConfig valueOf(String address, List<ProtocolModule> modules) {
public static ProviderConfig valueOf(String address, List<ProviderModule> 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<ProtocolModule> getModules() {
return modules;
public List<ProviderModule> getProviders() {
return providers;
}
public void setModules(List<ProtocolModule> modules) {
this.modules = modules;
public void setProviders(List<ProviderModule> 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);
}
}
net/src/main/java/com/zfoo/net/config/model/ProviderModule.java
+52
View File
@@ -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;
}
}
net/src/main/java/com/zfoo/net/consumer/Consumer.java
+46 -3
View File
@@ -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 <T extends IPacket> SyncAnswer<T> syncAsk(IPacket packet, Class<T> 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 <T extends IPacket> AsyncAnswer<T> asyncAsk(IPacket packet, Class<T> 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);
net/src/main/java/com/zfoo/net/consumer/balancer/AbstractConsumerLoadBalancer.java
+26 -9
View File
@@ -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<Session> sessionsByModule(ProtocolModule module) {
var clientSessionMap = NetContext.getSessionManager().getClientSessionMap();
var sessions = new ArrayList<Session>();
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);
}
}
net/src/main/java/com/zfoo/net/consumer/balancer/FixedConsumerLoadBalancer.java
-61
View File
@@ -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);
}
}
net/src/main/java/com/zfoo/net/consumer/registry/RegisterVO.java
+28 -14
View File
@@ -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<ProtocolModule> parseModules(String str) {
private static List<ProviderModule> 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<ConsumerModule> 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;
net/src/main/java/com/zfoo/net/schema/NetDefinitionParser.java
+28 -12
View File
@@ -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<BeanDefinitionHolder> parseModules(String param, Element element, ParserContext parserContext) {
var moduleElementList = DomUtils.getChildElementsByTagName(element, "module");
private ManagedList<BeanDefinitionHolder> parseProviderModules(String param, Element element, ParserContext parserContext) {
var moduleElementList = DomUtils.getChildElementsByTagName(element, "provider");
var providers = new ManagedList<BeanDefinitionHolder>();
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<BeanDefinitionHolder> parseConsumerModules(String param, Element element, ParserContext parserContext) {
var moduleElementList = DomUtils.getChildElementsByTagName(element, "consumer");
var modules = new ManagedList<BeanDefinitionHolder>();
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)));
}
net/src/main/resources/net-1.0.xsd
+13 -15
View File
@@ -23,32 +23,30 @@
<xsd:attribute name="password" type="xsd:string" use="required"/>
</xsd:complexType>
<xsd:complexType name="providerAttributeType">
<xsd:attribute name="name" type="xsd:string" use="required"/>
<xsd:attribute name="url" type="xsd:string" use="required"/>
</xsd:complexType>
<xsd:complexType name="providerType">
<xsd:complexType name="providersType">
<xsd:sequence>
<xsd:element name="module" maxOccurs="unbounded" type="moduleAttributeType" minOccurs="0"/>
<xsd:element name="provider" maxOccurs="unbounded" type="providerAttributeType" minOccurs="0"/>
</xsd:sequence>
<xsd:attribute name="task-dispatch" type="xsd:string" use="required"/>
<xsd:attribute name="thread" type="xsd:string" use="optional"/>
<xsd:attribute name="address" type="xsd:string" use="optional"/>
</xsd:complexType>
<xsd:complexType name="consumerType">
<xsd:complexType name="consumersType">
<xsd:sequence>
<xsd:element name="module" maxOccurs="unbounded" type="moduleAttributeType"/>
<xsd:element name="consumer" maxOccurs="unbounded" type="consumerAttributeType"/>
</xsd:sequence>
</xsd:complexType>
<xsd:complexType name="moduleAttributeType">
<xsd:attribute name="name" type="xsd:string" use="required"/>
<xsd:attribute name="load-balancer" type="xsd:string" default="consistent-hash"/>
<xsd:attribute name="group" type="xsd:string" default="0"/>
<xsd:complexType name="providerAttributeType">
<xsd:attribute name="provider" type="xsd:string" use="required"/>
<xsd:attribute name="protocol-module" type="xsd:string" use="required"/>
</xsd:complexType>
<xsd:complexType name="consumerAttributeType">
<xsd:attribute name="consumer" type="xsd:string" use="required"/>
<xsd:attribute name="load-balancer" type="xsd:string" default="consistent-hash"/>
</xsd:complexType>
<xsd:complexType name="configType">
<xsd:sequence>
@@ -59,10 +57,10 @@
<xsd:element name="monitor" type="addressType"/>
</xsd:choice>
<xsd:choice minOccurs="0" maxOccurs="1">
<xsd:element name="provider" type="providerType"/>
<xsd:element name="providers" type="providersType"/>
</xsd:choice>
<xsd:choice minOccurs="0" maxOccurs="1">
<xsd:element name="consumer" type="consumerType"/>
<xsd:element name="consumers" type="consumersType"/>
</xsd:choice>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:string" use="required"/>
net/src/test/java/com/zfoo/net/config/RegistryTest.java
-54
View File
@@ -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);
}
}
net/src/test/resources/provider/consumer_consistent_session_config.xml
+3 -4
View File
@@ -25,10 +25,9 @@
<net:address name="${registry.address.name}" url="${registry.address.url}"/>
</net:registry>
<net:consumer>
<net:module name="providerTest" load-balancer="consistent-hash"/>
</net:consumer>
<net:consumers>
<net:consumer consumer="providerTest" load-balancer="consistent-hash"/>
</net:consumers>
</net:config>
</beans>
net/src/test/resources/provider/consumer_fixed_config.xml
-34
View File
@@ -1,34 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:net="http://www.zfoo.com/schema/net"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-4.0.xsd
http://www.zfoo.com/schema/net
http://www.zfoo.com/schema/net-1.0.xsd">
<context:property-placeholder location="classpath:deploy-dev.properties"/>
<context:component-scan base-package="com.zfoo"/>
<net:config id="applicationNameTest" protocol-location="protocol.xml">
<net:registry center="${registry.center}" user="${registry.user}" password="${registry.password}">
<net:address name="${registry.address.name}" url="${registry.address.url}"/>
</net:registry>
<net:consumer>
<net:module name="providerTest" load-balancer="fixed"/>
</net:consumer>
</net:config>
</beans>
net/src/test/resources/provider/consumer_random_config.xml
+3 -4
View File
@@ -24,10 +24,9 @@
<net:address name="${registry.address.name}" url="${registry.address.url}"/>
</net:registry>
<net:consumer>
<net:module name="providerTest" load-balancer="random"/>
</net:consumer>
<net:consumers>
<net:consumer consumer="providerTest" load-balancer="random"/>
</net:consumers>
</net:config>
</beans>
net/src/test/resources/provider/consumer_shortest_time_config.xml
+3 -3
View File
@@ -25,9 +25,9 @@
</net:registry>
<net:consumer>
<net:module name="providerTest" load-balancer="shortest-time"/>
</net:consumer>
<net:consumers>
<net:consumer consumer="providerTest" load-balancer="shortest-time"/>
</net:consumers>
</net:config>
net/src/test/resources/provider/provider_config.xml
+3 -3
View File
@@ -25,9 +25,9 @@
<net:address name="${registry.address.name}" url="${registry.address.url}"/>
</net:registry>
<net:provider task-dispatch="consistent-hash">
<net:module name="providerTest"/>
</net:provider>
<net:providers task-dispatch="consistent-hash">
<net:provider protocol-module="providerTest" provider="providerTest"/>
</net:providers>
</net:config>
protocol/src/main/java/com/zfoo/protocol/registration/ProtocolModule.java
+2 -38
View File
@@ -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);
}
}