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

doc[module]: 深刻理解注册中心作用 1.自己是消费者则启动 2.自己是消费者，则连所有关心的服务提供者 3.节点变更影响服务提供者和消费者数量需检测 4.Client/ServerSessionMap内容变更

Browse Source
This commit is contained in:
jianan
2022-07-08 23:54:54 +08:00
parent d386884036
commit c2842801e1
5 changed files with 184 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
net/src/main/java/com/zfoo/net/config/manager/ConfigManager.java
+13 -1
View File
@@ -53,27 +53,36 @@ public class ConfigManager implements IConfigManager {
this.localConfig = localConfig;
}
/**
* 在调用这个方法之前：localConfig也就是NetConfig虽然已经被注册到Spring容器中，但是模块id等信息依然是没有的。这需要从protocol.xml中读取覆盖才行
*/
@Override
public void initRegistry() {
// 通过protocol，写入provider的module的id和version
var providerConfig = localConfig.getProvider();
if (Objects.nonNull(providerConfig) && CollectionUtils.isNotEmpty(providerConfig.getProviders())) {
// 检查并且替换配置文件中的ProtocolModule
// 服务提供者名字Set列表
var set = new HashSet<String>();
// 检查并且替换配置文件中的ProtocolModule
for (var providerModule : providerConfig.getProviders()) {
var provider = providerModule.getProvider();
var protocolModuleName = providerModule.getProtocolModule().getName();
// 从protocol中读值
var protocolModule = ProtocolManager.moduleByModuleName(protocolModuleName);
AssertionUtils.isTrue(protocolModule != null, "服务提供者[name:{}]在协议文件中不存在", protocolModuleName);
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);
@@ -83,6 +92,9 @@ public class ConfigManager implements IConfigManager {
AssertionUtils.isTrue(protocolModuleSet.add(protocolModule), "服务消费者[name:{}]重复消费了协议模块{}", consumer, protocolModule);
}
}
// 走到这之后，NetConfig通过app.xml(读取有哪些消费者)+protocol.xml(模块号信息)完成了初始化
// 接下来就是通过注册中心，把生产者和消费者关联起来
registry = new ZookeeperRegistry();
registry.start();
}
net/src/main/java/com/zfoo/net/config/model/NetConfig.java
+18 -3
View File
@@ -23,8 +23,6 @@ import java.util.Objects;
* @version 3.0
*/
public class NetConfig {
private String id;
private String protocolLocation;
@@ -34,7 +32,10 @@ public class NetConfig {
private boolean foldProtocol;
private String protocolPath;
private String protocolParam;
// 是否生成对应语言的协议
/**
* 是否生成对应语言的协议
*/
private boolean javascriptProtocol;
private boolean typescriptProtocol;
private boolean csharpProtocol;
@@ -43,10 +44,24 @@ public class NetConfig {
private boolean cppProtocol;
private boolean protobufProtocol;
/**
* 注册中心
*/
private RegistryConfig registry;
/**
* 监控
*/
private MonitorConfig monitor;
/**
* 生产者配置
*/
private ProviderConfig provider;
/**
* 消费者配置
*/
private ConsumerConfig consumer;
net/src/main/java/com/zfoo/net/consumer/registry/ZookeeperRegistry.java
+124 -4
View File
@@ -118,6 +118,16 @@ public class ZookeeperRegistry implements IRegistry {
*/
private final List<CuratorCache> listenerList = new ConcurrentArrayList<>();
/**
* 这个方法简单说就是分3大步骤：
* 如果配置中配置的有自己是服务提供者，那么自己作为服务提供者先启动 // 此时尚未注册到zk中
* 再启动zk // 干2件事：1.自己是服务提供者，将会注册到zk上 2.自己是服务消费者，会尝试创建TcpClient连接所有的自己要连接的provider
* 监听服务提供者节点的变更 // 自己作为消费者，服务提供者的变更会影响到自己
* <p>
* 总结：这个注册中心模块最终的结果我认为是，影响到了
* ClientSessionMap // 自己作为消费者去连接服务提供者连上后，存的session
* ServerSessionMap // 自己作为服务器角色，很多客户端连接自己，连上来后，保存到这
*/
@Override
public void start() {
var registryConfig = NetContext.getConfigManager().getLocalConfig().getRegistry();
@@ -126,26 +136,35 @@ public class ZookeeperRegistry implements IRegistry {
return;
}
// 先启动本地服务提供者，再启动curator
// 先启动本地服务提供者（服务提供者仅仅是一个TcpServer）
startProvider();
// 再启动curator框架 1.如果自己是服务提供者，将会注册自己到zk上 2.如果是服务消费者，会创建连接到自己关心的服务提供者上
startCurator();
// 检测服务提供者的增加或者减少。因为自己作为消费者的话，会影响到自己
startProviderCache();
}
private void startProvider() {
// 这个ProviderConfig包含了这个进程服务提供者信息
var providerConfig = NetContext.getConfigManager().getLocalConfig().getProvider();
// 这句意思是：提供了注册中心的配置(zk)，但是却没有服务提供者
if (Objects.isNull(providerConfig)) {
logger.info("服务提供者没有配置，不会在zk中注册服务，如果是单机启动请忽略这条日志");
return;
}
// 服务提供者也仅仅是一个TcpServer
// 这里可以看出并没有指定接口，是找一个可用的端口
var providerServer = new TcpServer(providerConfig.localHostAndPortOrDefault());
providerServer.start();
}
/**
* 启动curator框架，并且在zk客户端连接上zk服务器时：保证创建好/zfoo /provider /consumer 3个“持久化”节点
*/
private void startCurator() {
var registryConfig = NetContext.getConfigManager().getLocalConfig().getRegistry();
@@ -154,6 +173,7 @@ public class ZookeeperRegistry implements IRegistry {
.format("[center:{}]注册中心只能是zookeeper", JsonUtils.object2String(registryConfig)));
}
// 读取zk的配置，连接zk服务器
var zookeeperConnectStr = HostAndPort.toHostAndPortListStr(HostAndPort.toHostAndPortList(registryConfig.getAddress().values()));
var builder = CuratorFrameworkFactory.builder();
builder.connectString(zookeeperConnectStr);
@@ -169,19 +189,27 @@ public class ZookeeperRegistry implements IRegistry {
@Override
public void stateChanged(CuratorFramework client, ConnectionState state) {
switch (state) {
// zk客户端与zk服务器失去了连接(忽略此情况，使用本地配置的缓存)
case LOST:
// 忽略配置中心失去连接，使用本地配置的缓存
logger.error("[zookeeper:{}]失去连接，使用缓存", zookeeperConnectStr);
break;
// 暂停和只读这2种状态不检测
case SUSPENDED:
case READ_ONLY:
logger.warn("[zookeeper:{}]忽略的[state{}]", zookeeperConnectStr, state);
break;
// zk客户端和zk服务器重连了
case CONNECTED:
case RECONNECTED:
// 检查3个持久化节点，不存在就创建
createZookeeperRootPath();
// 如果自己是服务提供者，则注册自己
// 如果自己是消费者，则创建连接到所有的自己关心的服务提供者
initZookeeper();
break;
default:
logger.error("[zookeeper:{}]未知状态[state{}]", zookeeperConnectStr, state);
}
@@ -196,24 +224,34 @@ public class ZookeeperRegistry implements IRegistry {
}
}
/**
* 检查 /zfoo /zfoo/provider /zfoo/consumer 这3个“持久化”节点，不存在就创建
*/
private void createZookeeperRootPath() {
try {
// /zfoo
// 创建zookeeper的根路径
var rootStat = curator.checkExists().forPath(ROOT_PATH);
// 根节点不存在
if (Objects.isNull(rootStat)) {
var registryConfig = NetContext.getConfigManager().getLocalConfig().getRegistry();
var builder = curator.create();
builder.creatingParentsIfNeeded();
// 检查zk连接授权
if (registryConfig.hasZookeeperAuthor()) {
var zookeeperAuthorStr = registryConfig.toZookeeperAuthor();
var aclList = List.of(new ACL(ZooDefs.Perms.ALL, new Id("digest", DigestAuthenticationProvider.generateDigest(zookeeperAuthorStr))));
builder.withACL(aclList);
}
// 根节点是持久化节点
builder.withMode(CreateMode.PERSISTENT);
// 真正创建根节点
builder.forPath(ROOT_PATH, StringUtils.bytes(registryConfig.getCenter()));
} else {
var registryConfig = NetContext.getConfigManager().getLocalConfig().getRegistry();
// 读取根节点上的数据
var bytes = curator.getData().storingStatIn(new Stat()).forPath(ROOT_PATH);
// 把根节点数据从二进制转string字符串
var rootPathData = StringUtils.bytesToString(bytes);
// 检查zookeeper根节点的内容
@@ -237,7 +275,8 @@ public class ZookeeperRegistry implements IRegistry {
}
// /zfoo/provider
// 检查服务提供者节点，不存在则创建
var providerStat = curator.checkExists().forPath(PROVIDER_ROOT_PATH);
if (Objects.isNull(providerStat)) {
curator.create()
@@ -245,6 +284,8 @@ public class ZookeeperRegistry implements IRegistry {
.forPath(PROVIDER_ROOT_PATH, ArrayUtils.EMPTY_BYTE_ARRAY);
}
// /zfoo/consumer
// 检查消费者节点，不存在则创建
var consumerStat = curator.checkExists().forPath(CONSUMER_ROOT_PATH);
if (Objects.isNull(consumerStat)) {
curator.create()
@@ -273,10 +314,12 @@ public class ZookeeperRegistry implements IRegistry {
logger.error("不需要处理的[oldData:{}][newData:{}]", childDataToString(oldData), childDataToString(newData));
initZookeeper();
break;
case NODE_CREATED:
case NODE_CREATED: // 意味着有可能来了自己作为消费者需要关心的服务提供者
var providerStr = StringUtils.substringAfterFirst(newData.getPath(), PROVIDER_ROOT_PATH + StringUtils.SLASH);
var provider = RegisterVO.parseString(providerStr);
var localRegisterVO = NetContext.getConfigManager().getLocalConfig().toLocalRegisterVO();
// 如果启动的Consumer是自己关心的Consumer，那么就会接下来尝试连接他们
// 这意味着：如果有多个Consumer启动，那么最后将全部连接上去
if (RegisterVO.providerHasConsumer(provider, localRegisterVO)) {
providerHashConsumerSet.add(provider);
checkConsumer();
@@ -308,10 +351,14 @@ public class ZookeeperRegistry implements IRegistry {
private void initZookeeper() {
executor.execute(() -> {
try {
// 既有Provider，又有Consumer信息，注册到zk上，这是一个临时节点
// 这一步是：如果自己是服务提供者，就把自己注册上去
initLocalProvider();
// 自己是消费者，则连接所有自己关心的服务提供者
initConsumerCache();
} catch (Exception e) {
//
logger.error("zookeeper初始化失败，等待[{}]秒，重新初始化", RETRY_SECONDS, e);
SchedulerBus.schedule(new Runnable() {
@Override
@@ -323,12 +370,18 @@ public class ZookeeperRegistry implements IRegistry {
});
}
/**
* 如果自己是服务提供者，就讲自己注册上去
*
* @throws Exception
*/
private void initLocalProvider() throws Exception {
var localRegisterVO = NetContext.getConfigManager().getLocalConfig().toLocalRegisterVO();
if (Objects.nonNull(localRegisterVO.getProviderConfig())) {
var localProviderVoStr = localRegisterVO.toProviderString();
var localProviderPath = PROVIDER_ROOT_PATH + StringUtils.SLASH + localProviderVoStr;
// /zfoo/provider/tankHome | 192.168.3.2:12400 | provider:[3-tankHome-tankHomeProvider] | consumer:[4-tankCache-consistent-hash-tankCacheProvider]
var localProviderStat = curator.checkExists().forPath(localProviderPath);
if (Objects.isNull(localProviderStat)) {
curator.create()
@@ -352,23 +405,38 @@ public class ZookeeperRegistry implements IRegistry {
}
}
/**
* 遍历zk中的provider信息，从而找到所有自己关心的Provider，从而去连接他们
* 注意：自己作为Provider那自己启动下就行了比较简单。 但是作为Consumer，那么会尝试连接所有已经注册到zk上来的服务器信息
*
* @throws Exception
*/
private void initConsumerCache() throws Exception {
// tankHome | 192.168.3.2:12400 | provider:[3-tankHome-tankHomeProvider] | consumer:[4-tankCache-consistent-hash-tankCacheProvider]
var localRegisterVO = NetContext.getConfigManager().getLocalConfig().toLocalRegisterVO();
// 初始化providerCacheSet
// 遍历provider下注册的所有节点
var remoteProviderSet = curator.getChildren().forPath(PROVIDER_ROOT_PATH).stream()
.filter(it -> StringUtils.isNotBlank(it) && !"null".equals(it))
.map(it -> RegisterVO.parseString(it))
.filter(it -> Objects.nonNull(it))
// 检查是否这个节点是自己关心的节点
.filter(it -> RegisterVO.providerHasConsumer(it, localRegisterVO))
.collect(Collectors.toSet());
providerHashConsumerSet.clear();
// 将自己关心的节点存起来，接下来，将会开启TcpClient去连接这些Provider，连接上后，将会把这个session保存到ClientSessionMap中
providerHashConsumerSet.addAll(remoteProviderSet);
// 初始化consumer，providerCacheSet改变会导致消费者改变
// 如果自己没有连接上远程消费者，则会一直尝试连接
checkConsumer();
}
/**
* 不管是节点增删，都会调用，因为：有可能自己作为消费者，服务提供者增加了也可能减少了，都要检测下
*/
@Override
public void checkConsumer() {
if (curator == null) {
@@ -382,6 +450,9 @@ public class ZookeeperRegistry implements IRegistry {
executor.execute(() -> doCheckConsumer());
}
/**
* 检查下自己作为消费者，需要连接到的Provider是否全部连接上了，没连接上，就会创建TcpClient进行连接
*/
private void doCheckConsumer() {
if (curator.getState() != CuratorFrameworkState.STARTED) {
logger.error("curator还没有启动，忽略本次consumer的检查");
@@ -394,6 +465,7 @@ public class ZookeeperRegistry implements IRegistry {
for (var providerCache : providerHashConsumerSet) {
// 先排除已经启动的consumer
// getClientSessionMap
var consumerClientList = NetContext.getSessionManager().getClientSessionMap().values().stream()
.filter(it -> {
var attribute = it.getAttribute(AttributeType.CONSUMER);
@@ -416,12 +488,16 @@ public class ZookeeperRegistry implements IRegistry {
continue;
}
// 自己作为消费者，要创建一个TcpClient去连接服务提供者
var client = new TcpClient(HostAndPort.valueOf(providerCache.getProviderConfig().getAddress()));
var session = client.start();
// 自己作为消费者，使用TcpClient连接服务提供者不成功
if (Objects.isNull(session)) {
logger.error("[consumer:{}]启动失败，等待[{}]秒，重新检查consumer", providerCache, RETRY_SECONDS);
recheckFlag = true;
} else {
// 连接上了服务提供者
session.putAttribute(AttributeType.CONSUMER, providerCache);
EventBus.asyncSubmit(ConsumerStartEvent.valueOf(providerCache, session));
@@ -449,6 +525,13 @@ public class ZookeeperRegistry implements IRegistry {
}
}
/**
* 为某个路径下设置数据
*
* @param path
* @param bytes
* @param mode
*/
@Override
public void addData(String path, byte[] bytes, CreateMode mode) {
try {
@@ -468,6 +551,11 @@ public class ZookeeperRegistry implements IRegistry {
}
}
/**
* 删除路径
*
* @param path
*/
@Override
public void removeData(String path) {
try {
@@ -477,6 +565,12 @@ public class ZookeeperRegistry implements IRegistry {
}
}
/**
* 查询某个路径下的数据
*
* @param path
* @return
*/
@Override
public byte[] queryData(String path) {
try {
@@ -486,6 +580,12 @@ public class ZookeeperRegistry implements IRegistry {
}
}
/**
* 是否有某个路径
*
* @param path
* @return
*/
@Override
public boolean haveNode(String path) {
try {
@@ -495,6 +595,12 @@ public class ZookeeperRegistry implements IRegistry {
}
}
/**
* 查询某个路径下的所有子路径
*
* @param path
* @return
*/
@Override
public List<String> children(String path) {
try {
@@ -510,6 +616,11 @@ public class ZookeeperRegistry implements IRegistry {
return Collections.emptyList();
}
/**
* 查询所有服务信息(提供者+消费者信息)
*
* @return
*/
@Override
public Set<RegisterVO> remoteProviderRegisterSet() {
try {
@@ -527,6 +638,15 @@ public class ZookeeperRegistry implements IRegistry {
return Collections.emptySet();
}
/**
* 某个路径下发生数据变更（更新和创建）
* 数据删除
* 时，进行回调
*
* @param listenerPath 需要监听的路径
* @param updateCallback 回调方法，第一个参数是路径，第二个是变化的内容
* @param removeCallback 回调方法，第一个参数是路径，第二个是变化的内容
*/
@Override
public void addListener(String listenerPath, BiConsumer<String, byte[]> updateCallback, Consumer<String> removeCallback) {
try {
net/src/main/java/com/zfoo/net/schema/NetDefinitionParser.java
+25 -1
View File
@@ -84,12 +84,25 @@ public class NetDefinitionParser implements BeanDefinitionParser {
}
/**
* 解析config标签，并且注册NetConfig类到Spring容器
*
* @param element
* @param parserContext
*/
private void parseNetConfig(Element element, ParserContext parserContext) {
// 要注册到Spring的类，下面都是进行解析自定义标签，把值赋值给NetConfig的属性
var clazz = NetConfig.class;
var builder = BeanDefinitionBuilder.rootBeanDefinition(clazz);
// -----config属性解析-----
// 解析id
resolvePlaceholder("id", "id", builder, element, parserContext);
// 协议protocol.xml文件的位置。 注意：直接写protocol.xml 则是从resources目录下读
resolvePlaceholder("protocol-location", "protocolLocation", builder, element, parserContext);
// 各种语言的文件是否生成
resolvePlaceholder("javascript-protocol", "javascriptProtocol", builder, element, parserContext);
resolvePlaceholder("typescript-protocol", "typescriptProtocol", builder, element, parserContext);
resolvePlaceholder("csharp-protocol", "csharpProtocol", builder, element, parserContext);
@@ -97,26 +110,34 @@ public class NetDefinitionParser implements BeanDefinitionParser {
resolvePlaceholder("gdscript-protocol", "gdscriptProtocol", builder, element, parserContext);
resolvePlaceholder("cpp-protocol", "cppProtocol", builder, element, parserContext);
resolvePlaceholder("protobuf-protocol", "protobufProtocol", builder, element, parserContext);
// 文件是否折叠
resolvePlaceholder("fold-protocol", "foldProtocol", builder, element, parserContext);
resolvePlaceholder("protocol-path", "protocolPath", builder, element, parserContext);
resolvePlaceholder("protocol-param", "protocolParam", builder, element, parserContext);
// -----注册中心解析-----
// 上面解析的都是config标签的属性，这里开始解析registry元素
var registryElement = DomUtils.getFirstChildElementByTagName(element, "registry");
if (registryElement != null) {
parseRegistryConfig(registryElement, parserContext);
builder.addPropertyReference("registry", RegistryConfig.class.getCanonicalName());
}
// ----- monitor解析-----
var monitorElement = DomUtils.getFirstChildElementByTagName(element, "monitor");
if (monitorElement != null) {
parseMonitorConfig(monitorElement, parserContext);
builder.addPropertyReference("monitor", MonitorConfig.class.getCanonicalName());
}
// -----服务器生产者解析-----
var providerElement = DomUtils.getFirstChildElementByTagName(element, "providers");
if (providerElement != null) {
builder.addPropertyReference("provider", ProviderConfig.class.getCanonicalName());
parseProviderConfig(providerElement, parserContext);
builder.addPropertyReference("provider", ProviderConfig.class.getCanonicalName());
}
var consumerElement = DomUtils.getFirstChildElementByTagName(element, "consumers");
@@ -125,6 +146,7 @@ public class NetDefinitionParser implements BeanDefinitionParser {
builder.addPropertyReference("consumer", ConsumerConfig.class.getCanonicalName());
}
// 注册NetConfig到Spring容器中
parserContext.getRegistry().registerBeanDefinition(clazz.getCanonicalName(), builder.getBeanDefinition());
}
@@ -162,6 +184,8 @@ public class NetDefinitionParser implements BeanDefinitionParser {
var providerModules = parseProviderModules("providers", element, parserContext);
builder.addPropertyValue("providers", providerModules);
// 注册Consumer到Spring容器中
parserContext.getRegistry().registerBeanDefinition(clazz.getCanonicalName(), builder.getBeanDefinition());
}
net/src/main/java/com/zfoo/net/session/manager/SessionManager.java
+4 -1
View File
@@ -33,12 +33,15 @@ public class SessionManager implements ISessionManager {
/**
* 作为服务器，被别的客户端连接的Session
* 如：自己作为网关，那肯定有一大堆客户端连接，他们连接上来后，就会保存下来这些信息。
* 因此：要全局消息广播，其实要用这个Map
*/
private final Map<Long, Session> serverSessionMap = new ConcurrentHashMap<>();
/**
* 作为客户端，连接别的服务器的Session
* 作为客户端，连接别的服务器上后，保存下来的Session
* 如：自己配置了Consumer，说明自己作为消费者将要消费远程接口，就会创建一个TcpClient去连接Provider，那么连接上后，就会保存下来到这个Map中
*/
private final Map<Long, Session> clientSessionMap = new ConcurrentHashMap<>();