tiennm99/zfoo
1
0
Fork 0
mirror of https://github.com/tiennm99/zfoo.git synced 2026-08-11 16:26:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

ref[rpc]: refactor service provider

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