dubbo注册部分源码分析、集群策略、负载均衡算法
程序员文章站
2022-07-13 14:57:13
...
前面分别写了二篇文章，介绍dubbo的源码与模拟现场场景的结构与调用分析。目前还缺少注册与统计模块的分析，所以又抽空看了一下注册部分，越看越感觉dubbo真是宝啊，几乎全面的java知识都整合在一起了，看懂了DUBBO再看其它源代码都非常简单了；而且触类旁通，一些策略算法很多地方都可以用到。最后思考如何在特定场景下实现一个简单的soa的管理的minidubbo。
说到分析源码，我一般先网上找几篇看看，可是总看的云里雾里，而且dubbo升级后有些变化，所以还是自己硬啃一下。时间仓促欢迎斧正，我看的源码是2.5.3，我认为分析源码就不用贴多少代码了，关键理解其设计思路，解决问题需要的也是思路。有了思路就是COPY，修改，调试了。
一、dubbo注册分析
网上的源码最大的问题是没有一个完整的思维模型引导，所以我先介绍我所了解的注册部分的思维模型。注册就相当于一个购物平台，总不能买卖双方总是直接调用吧，所以可以这么理解：调用方如果配置注册中心，就注册自己，并从平台查询，可以得到自己想到的实现服务的被调用方地址，这个是订阅过程，会不断返回一些被调用方的URL，而这些个URL就是真正的被调用方了，存起来并按一定负载规则选择一个当成直接调用就OK了。而被调用方也是注册自己，并从平台得到..些URL重新暴露。（这里没明白为要得到那些URL并重新doChangeLocalExport，反正也是订阅并传入一个监听器，但不影响整体理解）。而注册时的数据存放可以是zookeeper,radis，也可以是数据库，但注册，订阅等业务逻辑都是调用端与被调用端来实现的。
调用方的一个接口，从注册中心找到多个实现
当调用端直接配置被调用端时，用dubboProtocol的refer方法就生成invoker,而当配置为注册中心时，调用方的接口就要按registryProtocol提供的refer生成invoker,而且一个接口配置了多个注册中心，并且有多个提供方时，那就比较麻烦了。我买一件商品，那么多商家提供，那当然弱水三千只取一瓢了，但可以随机，也可以选择最便宜的。
registryProtocol与RegistryDirectory是两个核心的业务逻辑类，另外根据注册中心的不同还有些不同的注册器与注册器工厂，比如zookeeper，radis的。
我们看到，registryProtocol做refer的时候得到相应的注册器，这里是比较复杂的SPI方式，后面会详细介绍的。比如得到zookeeper的，再做dorefer的时候三件事，先是注册registry.register(*)，再是订阅，订阅就是我扔个回调对象给你，你有事了通知回调对象就行了，我就知道了。代码就是directory.subscribe(*）内部是registry.subscribe(url, this);。最后就是cluster.join(directory);
总结一下，得到注册中心的地址后，调用方用zookeeper注册器来注册一下，并传入一个对象RegistryDirectory向注册器订阅可用的url，注册器会有一个线程不断的查找zookeeper的节点得到可用的url来通知RegistryDirectory。RegistryDirectory拿到这些url就一个个得到invokers,这些invokers的持有者RegistryDirectory会生成一个实现了invoker的代理对象。而客户端以后就调用这个invoker的代理对象了，代理对象会用策略从一堆真正的invoker中拿一个来用的。
----------------------《把URL转成invoker代码》--------------------
拿到一个注册中心的真正一堆url后，转化成一堆invoker的重要语句如下：
//把订阅的url转化为invoker就是RegistryDirectory类中的refreshInvoker方法中的这句
Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls) ;
//转化过程就是用注册中心告诉你的真正的一堆url,用各自的协议来refer一下，生成一堆invoker。
invoker = new InvokerDelegete<T>(protocol.refer(serviceType, url), url, providerUrl);
invoker代理对象找出一个invoker来用就是FailoverClusterInvoker中的下面这句：
Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
----------------------------《END》------------------------
疑问：有几个注册中心，如果每个中心都查到有实现，那就有几个代理对象吗？那又多一层选择吗？一个调用先选择一个代理对象，代理对象中又选择出一个真正的invoker?还是把这些全部合并到一起了？这个细节还没有看。发现ReferenceConfig中有：invoker = cluster.join(new StaticDirectory(u, invokers));也是集群策略。
解答：今天又看了一会代码，明白了。当我们从afterPropertiesSet()开始追踪调用端对象的诞生时，直到看这个方法createProxy(Map<String, String> map)，最终生成调用端的代理对象。
-------------《多个注册中心返回的invokers集群代码》-----------
//加载所有的注册中心地址
List<URL> us = loadRegistries(false);
...
for (URL url : urls) {
//每一个注册中心生成一个invoker，放在一起。注意一个注册中心生成的是一个集群failover的invoker。
invokers.add(refprotocol.refer(interfaceClass, url));
}
...
//把每个注册中心的invoker再组成一个集群。这里不是failoverCluster了，是AvailableCluster了。
invoker = cluster.join(new StaticDirectory(u, invokers));
------------------------------《END》-------------------------
就是说：一个注册中心得到的是集群，而多个注册中心是集群的集群。但是集群策略不一样。其实之前已猜测到了，只是对refprotocol有点困惑，所有的注册中心怎么用同一个变量refprotocol来refer呢?不同的注册中心协议不一样啊，有zookeeper,radis等。当然我知道是spi进来的，但总不能用一个变量吧。突然想起之前的文章中我还分析过spi的特别之处。阿里的spi得到的是一个适配器，是用生成代码编译方式的，具体适配是那个类，要根据参数定的。那就对了，注册中心的URL是不一样的，所以得到不同的类。
此刻又一想，注册协议只有一个RegistryProtocol，不同的注册中心导致注册协议refer时用的注册器不一样，那是如何获取不同的注册器呢？RegistryProtocol中有一个属性是：RegistryFactory registryFactory;它有Set方法。我又回看了一下与SPI有关的ExtensionLoader类中有injectExtension()，对所有SPI加载进来的再注入一下，那注入的内容也可以又是SPI加载的东西。那一切都清楚了：RegistryProtocol是protocol.class通过SPI加载的，而RegistryProtocol内部的RegistryFactory 又是通过SPI加载并注入的。而SPI注入的都是适配器类，代理了相关的实际操作类，具体代理谁是通过传入参数临时决定的，决定的时候再加载真正的扩展类（SPI或者SPING方式）。而参数都是URL中，所以URL决定了真正用什么协议，以及真正用什么注册器工厂。工厂确定了，那注册类就确定了。一切在运行时都搞定了。
protocol的SPI加载时是适配器，运行时如果恰好是RegistryProtocol，那它内部还有RegistryFactory、ProxyFactory也是SPI加载的适配器，具体运行时如果是DubboRegistryFactory ，它的内部又有Cluster是SPI的，如果运行时恰好是....那它内部.. 真是层层注入的加载类，实现的是一个微核心功能，太有创意了！其实sping的IOC容器不也是这样层层注入的嘛，可能DUBBO“抄袭”的就是这样的思路吧。真是忙了几天别的事，思维有点断，还好理顺了！
二、注册中的集群策略与负载均衡算法汇总与应用场景
DUBBO中的宝贝越挖越多，这也是发现的一批重要知识点，在分布式环境中用的非常多，都总结到一起了，可以对比学习使用。
1.负载均衡算法--只要有集群都需要，在电商业务中，比如对不同级别的用户，权重也可能有差别，很多地方用的到。
1.1 RandomLoadBalance：
如果权重一样，就是简单的random，如果有权重，那把权重累加起来，从中得到一个随机数，在循环中不断累减随机数，如果出现负数就是要的值。这个算法没有去找证明，但我相信没问题，碰到了直接用。
1.2 ConsistentHashLoadBalance：
利用TreeMap实现的一致性HASH算法
1.3 RoundRobinLoadBalance：即轮询调度算法。比如有ABCD四个invoker，依次调用ABCD ABCD ABCD ABCD A...。如果加上权重稍微复杂一点，比如权重AC是2，BD是1，那次序是：ABCDAC ABCDAC ABCDAC.....。
2.集群策略
2.1 AvailableCluster:这个简单，从invokers中随便拿一个isAvailable()的invoker.拿的时候不需要负载均衡策略。
2.2 FailoverCluster：从invokers中按照负载均衡策略拿出来，可用就OK。不可用的话循环拿其它没拿过的。通用用于读操作，这个失败了换下一个。
FailfastCluster：从invokers中按照负载均衡策略拿出来一个，就用它了。不行就抛出异常。常用于非幂等性的写操作。非幂等就是一次操作与多次操作结果不一样。
2.3 FailbackCluster：从invokers中按照负载均衡策略拿出来一个，如果执行不成功，那记录下来，产生一个定时线程不断的重试。一般用于消息通知。
2.4 ForkingCluster：从invokers中按照负载均衡策略拿出来多个来,谁先有结果就是谁，用线程池来执行，把结果写入一个LinkedBlockingQueue，如果都是异常，把最后一个异常写入。主线程等待一定时间后用poll来取第一个。
这是并行调用，只要一个成功即返回，通常用于实时性要求较高的操作，但需要浪费更多服务资源。
2.5 FailsafeCluster：有异常时，也返回一个正常的空对象new RpcResult()。这个是失败安全，出现异常时，直接忽略，通常用于写入审计日志等操作。
2.6 BroadcastCluster：广播方式，就是从invokers中循环执行所有的invoker。大家一个接一个都来做一遍调用。
三、重点分析下在负载均衡中用treemap实现的一致性Hash算法：
一致性hash在分布式环节下非常重要的选择方式。比如自己开发redis分布式中间件，比如分表分库等操作中。都需要命中目标并且目标个数是可以扩展的。
com.alibaba.dubbo.rpc.cluster.loadbalance.ConsistentHashLoadBalance中有一个内部静态类ConsistentHashSelector。
3.1 属性如下：
private final TreeMap<Long, Invoker<T>> virtualInvokers;
private final int replicaNumber;
private final int identityHashCode;
private final int[] argumentIndex;
3.2 构造时方法如下：
for (Invoker<T> invoker : invokers) {
for (int i = 0; i < replicaNumber / 4; i++) {
byte[] digest = md5(invoker.getUrl().toFullString() + i);
for (int h = 0; h < 4; h++) {
long m = hash(digest, h);
virtualInvokers.put(m, invoker);
}
}
}
说明：treemap里放的是一些invoker，key是一个hash码，是long类型。比如有10个invorker，复制数为5，那一共放了50个key,value在treemap中。每5个key对应的invorker是一样的。一般想的是有几个invoker就放入tree中几个，但实际上为了均匀命中率会都多复制。
3.3 如何选择出invoker?
public Invoker<T> select(Invocation invocation) {
String key = toKey(invocation.getArguments());
byte[] digest = md5(key);
Invoker<T> invoker = sekectForKey(hash(digest, 0));
return invoker;
}
private Invoker<T> sekectForKey(long hash) {
Invoker<T> invoker;
Long key = hash;
if (!virtualInvokers.containsKey(key)) {
SortedMap<Long, Invoker<T>> tailMap = virtualInvokers.tailMap(key);
if (tailMap.isEmpty()) {
key = virtualInvokers.firstKey();
} else {
key = tailMap.firstKey();
}
}
invoker = virtualInvokers.get(key);
return invoker;
}
说明：传入的是一个invocation，那怎么找到invoker呢？首先invocation得到key，key再得到digest，digest 再得到hash值。重点来了，就是如何用这个hash值在treemap上找到特定的invoker。首先对原码中的select写成sekect的拼写错误不习惯。
根据上面得到的 hash值key，得到一个 tailMap = virtualInvokers.tailMap(key); tailMap是一个大于等于key的一部分树的数据，等于的几率很小，一般是比key大的，那取这个tailMap中的第一个，就等于取了比key大的之中最小的一个tree中存在的有效key了。当然也有可能取不到，tailMap为空，因为tree中最大的有效key也比参数key小，那当然返回treeMap中最小的喽，这样不就是一个接好的圆环了。
3.4 说说hash算法
有些md5，sha1之类的方法得到hash值。还有些位运算的处理，就不深入研究了。
说到分析源码，我一般先网上找几篇看看，可是总看的云里雾里，而且dubbo升级后有些变化，所以还是自己硬啃一下。时间仓促欢迎斧正，我看的源码是2.5.3，我认为分析源码就不用贴多少代码了，关键理解其设计思路，解决问题需要的也是思路。有了思路就是COPY，修改，调试了。
一、dubbo注册分析
网上的源码最大的问题是没有一个完整的思维模型引导，所以我先介绍我所了解的注册部分的思维模型。注册就相当于一个购物平台，总不能买卖双方总是直接调用吧，所以可以这么理解：调用方如果配置注册中心，就注册自己，并从平台查询，可以得到自己想到的实现服务的被调用方地址，这个是订阅过程，会不断返回一些被调用方的URL，而这些个URL就是真正的被调用方了，存起来并按一定负载规则选择一个当成直接调用就OK了。而被调用方也是注册自己，并从平台得到..些URL重新暴露。（这里没明白为要得到那些URL并重新doChangeLocalExport，反正也是订阅并传入一个监听器，但不影响整体理解）。而注册时的数据存放可以是zookeeper,radis，也可以是数据库，但注册，订阅等业务逻辑都是调用端与被调用端来实现的。
调用方的一个接口，从注册中心找到多个实现
当调用端直接配置被调用端时，用dubboProtocol的refer方法就生成invoker,而当配置为注册中心时，调用方的接口就要按registryProtocol提供的refer生成invoker,而且一个接口配置了多个注册中心，并且有多个提供方时，那就比较麻烦了。我买一件商品，那么多商家提供，那当然弱水三千只取一瓢了，但可以随机，也可以选择最便宜的。
registryProtocol与RegistryDirectory是两个核心的业务逻辑类，另外根据注册中心的不同还有些不同的注册器与注册器工厂，比如zookeeper，radis的。
我们看到，registryProtocol做refer的时候得到相应的注册器，这里是比较复杂的SPI方式，后面会详细介绍的。比如得到zookeeper的，再做dorefer的时候三件事，先是注册registry.register(*)，再是订阅，订阅就是我扔个回调对象给你，你有事了通知回调对象就行了，我就知道了。代码就是directory.subscribe(*）内部是registry.subscribe(url, this);。最后就是cluster.join(directory);
总结一下，得到注册中心的地址后，调用方用zookeeper注册器来注册一下，并传入一个对象RegistryDirectory向注册器订阅可用的url，注册器会有一个线程不断的查找zookeeper的节点得到可用的url来通知RegistryDirectory。RegistryDirectory拿到这些url就一个个得到invokers,这些invokers的持有者RegistryDirectory会生成一个实现了invoker的代理对象。而客户端以后就调用这个invoker的代理对象了，代理对象会用策略从一堆真正的invoker中拿一个来用的。
----------------------《把URL转成invoker代码》--------------------
拿到一个注册中心的真正一堆url后，转化成一堆invoker的重要语句如下：
//把订阅的url转化为invoker就是RegistryDirectory类中的refreshInvoker方法中的这句
Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls) ;
//转化过程就是用注册中心告诉你的真正的一堆url,用各自的协议来refer一下，生成一堆invoker。
invoker = new InvokerDelegete<T>(protocol.refer(serviceType, url), url, providerUrl);
invoker代理对象找出一个invoker来用就是FailoverClusterInvoker中的下面这句：
Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
----------------------------《END》------------------------
疑问：有几个注册中心，如果每个中心都查到有实现，那就有几个代理对象吗？那又多一层选择吗？一个调用先选择一个代理对象，代理对象中又选择出一个真正的invoker?还是把这些全部合并到一起了？这个细节还没有看。发现ReferenceConfig中有：invoker = cluster.join(new StaticDirectory(u, invokers));也是集群策略。
解答：今天又看了一会代码，明白了。当我们从afterPropertiesSet()开始追踪调用端对象的诞生时，直到看这个方法createProxy(Map<String, String> map)，最终生成调用端的代理对象。
-------------《多个注册中心返回的invokers集群代码》-----------
//加载所有的注册中心地址
List<URL> us = loadRegistries(false);
...
for (URL url : urls) {
//每一个注册中心生成一个invoker，放在一起。注意一个注册中心生成的是一个集群failover的invoker。
invokers.add(refprotocol.refer(interfaceClass, url));
}
...
//把每个注册中心的invoker再组成一个集群。这里不是failoverCluster了，是AvailableCluster了。
invoker = cluster.join(new StaticDirectory(u, invokers));
------------------------------《END》-------------------------
就是说：一个注册中心得到的是集群，而多个注册中心是集群的集群。但是集群策略不一样。其实之前已猜测到了，只是对refprotocol有点困惑，所有的注册中心怎么用同一个变量refprotocol来refer呢?不同的注册中心协议不一样啊，有zookeeper,radis等。当然我知道是spi进来的，但总不能用一个变量吧。突然想起之前的文章中我还分析过spi的特别之处。阿里的spi得到的是一个适配器，是用生成代码编译方式的，具体适配是那个类，要根据参数定的。那就对了，注册中心的URL是不一样的，所以得到不同的类。
此刻又一想，注册协议只有一个RegistryProtocol，不同的注册中心导致注册协议refer时用的注册器不一样，那是如何获取不同的注册器呢？RegistryProtocol中有一个属性是：RegistryFactory registryFactory;它有Set方法。我又回看了一下与SPI有关的ExtensionLoader类中有injectExtension()，对所有SPI加载进来的再注入一下，那注入的内容也可以又是SPI加载的东西。那一切都清楚了：RegistryProtocol是protocol.class通过SPI加载的，而RegistryProtocol内部的RegistryFactory 又是通过SPI加载并注入的。而SPI注入的都是适配器类，代理了相关的实际操作类，具体代理谁是通过传入参数临时决定的，决定的时候再加载真正的扩展类（SPI或者SPING方式）。而参数都是URL中，所以URL决定了真正用什么协议，以及真正用什么注册器工厂。工厂确定了，那注册类就确定了。一切在运行时都搞定了。
protocol的SPI加载时是适配器，运行时如果恰好是RegistryProtocol，那它内部还有RegistryFactory、ProxyFactory也是SPI加载的适配器，具体运行时如果是DubboRegistryFactory ，它的内部又有Cluster是SPI的，如果运行时恰好是....那它内部.. 真是层层注入的加载类，实现的是一个微核心功能，太有创意了！其实sping的IOC容器不也是这样层层注入的嘛，可能DUBBO“抄袭”的就是这样的思路吧。真是忙了几天别的事，思维有点断，还好理顺了！
二、注册中的集群策略与负载均衡算法汇总与应用场景
DUBBO中的宝贝越挖越多，这也是发现的一批重要知识点，在分布式环境中用的非常多，都总结到一起了，可以对比学习使用。
1.负载均衡算法--只要有集群都需要，在电商业务中，比如对不同级别的用户，权重也可能有差别，很多地方用的到。
1.1 RandomLoadBalance：
如果权重一样，就是简单的random，如果有权重，那把权重累加起来，从中得到一个随机数，在循环中不断累减随机数，如果出现负数就是要的值。这个算法没有去找证明，但我相信没问题，碰到了直接用。
1.2 ConsistentHashLoadBalance：
利用TreeMap实现的一致性HASH算法
1.3 RoundRobinLoadBalance：即轮询调度算法。比如有ABCD四个invoker，依次调用ABCD ABCD ABCD ABCD A...。如果加上权重稍微复杂一点，比如权重AC是2，BD是1，那次序是：ABCDAC ABCDAC ABCDAC.....。
2.集群策略
2.1 AvailableCluster:这个简单，从invokers中随便拿一个isAvailable()的invoker.拿的时候不需要负载均衡策略。
2.2 FailoverCluster：从invokers中按照负载均衡策略拿出来，可用就OK。不可用的话循环拿其它没拿过的。通用用于读操作，这个失败了换下一个。
FailfastCluster：从invokers中按照负载均衡策略拿出来一个，就用它了。不行就抛出异常。常用于非幂等性的写操作。非幂等就是一次操作与多次操作结果不一样。
2.3 FailbackCluster：从invokers中按照负载均衡策略拿出来一个，如果执行不成功，那记录下来，产生一个定时线程不断的重试。一般用于消息通知。
2.4 ForkingCluster：从invokers中按照负载均衡策略拿出来多个来,谁先有结果就是谁，用线程池来执行，把结果写入一个LinkedBlockingQueue，如果都是异常，把最后一个异常写入。主线程等待一定时间后用poll来取第一个。
这是并行调用，只要一个成功即返回，通常用于实时性要求较高的操作，但需要浪费更多服务资源。
2.5 FailsafeCluster：有异常时，也返回一个正常的空对象new RpcResult()。这个是失败安全，出现异常时，直接忽略，通常用于写入审计日志等操作。
2.6 BroadcastCluster：广播方式，就是从invokers中循环执行所有的invoker。大家一个接一个都来做一遍调用。
三、重点分析下在负载均衡中用treemap实现的一致性Hash算法：
一致性hash在分布式环节下非常重要的选择方式。比如自己开发redis分布式中间件，比如分表分库等操作中。都需要命中目标并且目标个数是可以扩展的。
com.alibaba.dubbo.rpc.cluster.loadbalance.ConsistentHashLoadBalance中有一个内部静态类ConsistentHashSelector。
3.1 属性如下：
private final TreeMap<Long, Invoker<T>> virtualInvokers;
private final int replicaNumber;
private final int identityHashCode;
private final int[] argumentIndex;
3.2 构造时方法如下：
for (Invoker<T> invoker : invokers) {
for (int i = 0; i < replicaNumber / 4; i++) {
byte[] digest = md5(invoker.getUrl().toFullString() + i);
for (int h = 0; h < 4; h++) {
long m = hash(digest, h);
virtualInvokers.put(m, invoker);
}
}
}
说明：treemap里放的是一些invoker，key是一个hash码，是long类型。比如有10个invorker，复制数为5，那一共放了50个key,value在treemap中。每5个key对应的invorker是一样的。一般想的是有几个invoker就放入tree中几个，但实际上为了均匀命中率会都多复制。
3.3 如何选择出invoker?
public Invoker<T> select(Invocation invocation) {
String key = toKey(invocation.getArguments());
byte[] digest = md5(key);
Invoker<T> invoker = sekectForKey(hash(digest, 0));
return invoker;
}
private Invoker<T> sekectForKey(long hash) {
Invoker<T> invoker;
Long key = hash;
if (!virtualInvokers.containsKey(key)) {
SortedMap<Long, Invoker<T>> tailMap = virtualInvokers.tailMap(key);
if (tailMap.isEmpty()) {
key = virtualInvokers.firstKey();
} else {
key = tailMap.firstKey();
}
}
invoker = virtualInvokers.get(key);
return invoker;
}
说明：传入的是一个invocation，那怎么找到invoker呢？首先invocation得到key，key再得到digest，digest 再得到hash值。重点来了，就是如何用这个hash值在treemap上找到特定的invoker。首先对原码中的select写成sekect的拼写错误不习惯。
根据上面得到的 hash值key，得到一个 tailMap = virtualInvokers.tailMap(key); tailMap是一个大于等于key的一部分树的数据，等于的几率很小，一般是比key大的，那取这个tailMap中的第一个，就等于取了比key大的之中最小的一个tree中存在的有效key了。当然也有可能取不到，tailMap为空，因为tree中最大的有效key也比参数key小，那当然返回treeMap中最小的喽，这样不就是一个接好的圆环了。
3.4 说说hash算法
有些md5，sha1之类的方法得到hash值。还有些位运算的处理，就不深入研究了。
上一篇： 优秀文章