tiennm99/zfoo
1
0
Fork 0
mirror of https://github.com/tiennm99/zfoo.git synced 2026-08-05 22:25:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf[net]: 重构了net的服务提供者和服务消费者的配置方式

Browse Source
This commit is contained in:
jaysunxiao
2022-07-01 23:33:48 +08:00
parent 06155340f8
commit ef66cfbc3a
14 changed files with 118 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
net/src/main/java/com/zfoo/net/config/manager/ConfigManager.java
+21 -10
View File
@@ -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<String>();
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<String>();
var protocolModuleSet = new HashSet<ProtocolModule>();
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;
}
}
net/src/main/java/com/zfoo/net/config/model/ConsumerModule.java
+39 -3
View File
@@ -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);
}
}
net/src/main/java/com/zfoo/net/config/model/ProviderModule.java
+25 -6
View File
@@ -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);
}
}
net/src/main/java/com/zfoo/net/consumer/Consumer.java
+5 -23
View File
@@ -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;
net/src/main/java/com/zfoo/net/consumer/IConsumer.java
+4
View File
@@ -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);
/**
* 直接发送，不需要任何返回值
*
net/src/main/java/com/zfoo/net/consumer/registry/RegisterVO.java
+13 -10
View File
@@ -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())));
net/src/main/java/com/zfoo/net/consumer/registry/ZookeeperRegistry.java
+1 -1
View File
@@ -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;
net/src/main/java/com/zfoo/net/schema/NetDefinitionParser.java
+3 -2
View File
@@ -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)));
}
net/src/main/resources/net-1.0.xsd
+2 -1
View File
@@ -44,8 +44,9 @@
</xsd:complexType>
<xsd:complexType name="consumerAttributeType">
<xsd:attribute name="consumer" type="xsd:string" use="required"/>
<xsd:attribute name="protocol-module" type="xsd:string" use="required"/>
<xsd:attribute name="load-balancer" type="xsd:string" default="consistent-hash"/>
<xsd:attribute name="consumer" type="xsd:string" use="required"/>
</xsd:complexType>
<xsd:complexType name="configType">
net/src/test/java/com/zfoo/net/core/provider/ProviderTest.java
-21
View File
@@ -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);
}
}
net/src/test/resources/provider/consumer_consistent_session_config.xml
+1 -2
View File
@@ -20,13 +20,12 @@
<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:consumers>
<net:consumer consumer="providerTest" load-balancer="consistent-hash"/>
<net:consumer protocol-module="providerTest" load-balancer="consistent-hash" consumer="myProvider1"/>
</net:consumers>
</net:config>
net/src/test/resources/provider/consumer_random_config.xml
+1 -2
View File
@@ -19,13 +19,12 @@
<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:consumers>
<net:consumer consumer="providerTest" load-balancer="random"/>
<net:consumer protocol-module="providerTest" load-balancer="random" consumer="myProvider1"/>
</net:consumers>
</net:config>
net/src/test/resources/provider/consumer_shortest_time_config.xml
+1 -4
View File
@@ -19,16 +19,13 @@
<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:consumers>
<net:consumer consumer="providerTest" load-balancer="shortest-time"/>
<net:consumer protocol-module="providerTest" load-balancer="shortest-time" consumer="myProvider1"/>
</net:consumers>
</net:config>
</beans>
net/src/test/resources/provider/provider_config.xml
+2 -3
View File
@@ -20,15 +20,14 @@
<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:providers task-dispatch="consistent-hash">
<net:provider protocol-module="providerTest" provider="providerTest"/>
<net:provider protocol-module="providerTest" provider="myProvider1"/>
<net:provider protocol-module="providerTest" provider="myProvider2"/>
</net:providers>
</net:config>
</beans>