Dubbo(原理浅析篇)

柔光的暖阳◎ 2023-07-18 09:46 4阅读 0赞

#### dubbo spi扩展 ####

dubbo扩展利用java的spi机制，使高层引用底层，用户可以通过spi扩展dubbo。  
**java spi**  
通过在META-INF目录创建`services`文件夹，然后以`接口全限定名`作为文件名，多个`实现类的权限定名`为文件内容，通过`ServiceLoader`类的load方法传入接口class，即可获取到所有实现类。原理是通过在指定目录查找该类的文件，从文件中获取权限定名并创建对象。  
**dubbo**  
dubbo没有直接使用java spi，而是实现一套类似的机制，增加了一些功能。  
通过在META-INF目录`dubbo`文件夹存放配置文件，文件名命名方式与java spi一样，不同的是，文件内容以`键值对`的方式，可以对象名为key，value为实现类的全限定名。通过`ExtensionLoader`可以获取实现类，并且可以通过key直接获取对应的实现对象。

源码在ExtensionLoader，主要就是现将指定目录配置的所有扩展类通过反射实例化，然后放入通过key value形势存到缓存中。通过指定获取的扩展类key，获取到对应实现类。之后为实现类设置依赖，通过实例中的set方法进行设置，这是dubbo的ioc机制

#### 服务导出 ####

服务通过注册spring上下文刷新事件，在spring初始化会开始导出服务。  
源码`ServiceBean`。  
获取配置信息，通过解析xml，开始服务导出。

1.  通过`NettyServer`启动一个服务，创建Netty包下的ServerBootstrap启动服务。底层基于nio的`ServerSocketChannel`启动一个服务，监听端口。
2.  向注册中心注册服务。  
    源码在ZookeeperRegistry，先根据zk配置信息创建zk连接客户端，再以 分组名/接口全限定名/provider/$\{url\} 目录下创建一个临时节点，即创建一个文件代表一条注册信息。默认分组名为`dubbo`。

#### 服务引用 ####

Dubbo 服务引用的时机有两个，

1.  Spring 容器调用 ReferenceBean 的 afterPropertiesSet 方法时引用服务
2.  ReferenceBean 对应的服务被注入到其他类中时引用。默认是第二种，用到才注入。

在类ReferenceBean中，引用bean通过实现spring的FactoryBean的getObject方法，定义生成引用bean的方式，该方法最终调用了`ReferenceConfig`的init方法，该方法先生成消费者配置类，然后判断是jvm本地引用还是远程引用，如果`有直接连接服务的url或者url参数是远程作用域、本地导出的服务没有包含当前服务`等等情况则为远程。

如果url只有一个地址，则直接生成`invoke`对象。invoke调用通过`NettyChannel`发送消息，通过ChannelFuture获取调用结果。

如果有多个url则会有多个invoker对象封装成`cluster`对象，获取到一个合并后的inoker对象。最后生成引用类代理对象，如果不是生成代理对象，则客户端需要直接依赖invoker对象，造成代码侵入。  
dubbo默认生成代理通过`javassistProxy`,好处时执行代理方法比jdk快，但是生成代理类比jdk慢.  
invoke调用服务默认时通过`netty`通信。

通过配置可以修改代理方式为jdk  
<dubbo:provider proxy=“jdk” />

#### xml解析，将bean注入spring ####

源码: DubboBeanDefinitionParser.parse()  
为bean生成BeanDefinition注册到spring中，引用bean会实现factoryBean,调用bean时会通过getObject获取实际的bean,getObject会返回一个代理对象，由代理对象与服务端进行网络通信获取处理结果。

#### 集群容错 ####

##### 服务目录 #####

根据服务提供者配置生成的invoke对象集合，而且集合的内容会跟随注册中心服务的变化而变化。源码在`RegistryDirectory`.

##### 服务路由 #####

条件路由，决定了消费端的调用目标。通过配置路由规则，可以限制某些服务提供者只可以被某些服务消费者调用。相当于服务网关的。源码在`ConditionRouter`,对配置进行解析匹配。

##### 集群 #####

由于服务提供者大部分情况是会部署多台机器，dubbo通过封装Cluster接口，将多个invoke合并成一个，消费端只需要通过这个inovke进行调用，不用底层具体调用了哪个api

##### 集群容错 #####

**Failover Cluster - 失败自动切换**  
FailoverClusterInvoker。失败时，获取允许重试次数，然后循环重试，每次重试会重新获取invoke集合，线通过负载均衡选出一个invoke，但是如果选出的invoke不稳定（当前调用失败过），则会选择下一个invoke.会判断是否是粘滞连接，是的话默认会同个消费者会一直调用同一个提供者，但是如果是已经执行失败过，重试则不会再调用同一个提供者。  
**Failfast Cluster - 快速失败**  
FailfastClusterInvoker。调用一次，失败就抛异常  
**Failsafe Cluster - 失败安全**  
FailsafeClusterInvoker。调用一次，失败打印日志  
**Failback Cluster - 失败自动恢复**  
FailbackClusterInvoker。调用失败打印日志，记录到失败任务中，通过synchronized开启同步，创建定时任务，每5秒扫一次失败任务进行遍历重试。  
**Forking Cluster - 并行调用多个服务提供者**  
ForkingClusterInvoker。遍历invoke集合，通过`线程池`每个inovke一个线程`异步`执行，拿到返回结果就将结果放入阻塞队列。主线程从阻塞队列拿任务，只要阻塞队列有一个执行完成拿到结果就返回（poll自动阻塞等待队列有结果）。需要注意的是，任一个invoke执行会进行计数，只有失败量等于invoke数，才会把异常放入队列，保证只要有一个成功即返回成功。  
**Broadcast Cluster**  
同步遍历执行，如果有一个异常则暂存异常对象，遍历执行结束后，判断存在异常则抛出异常。（广播处理，可以用于更新提供本地缓存）

##### 整体流程 #####

*  服务初始化阶段

为服务消费者创建cluster invoker

*  服务调用阶段

整个过程先从服务目录获取服务列表，再经过服务路由，筛选符合路由条件的服务(如果没有配置则无限制)，再经过负载均衡策略选择某一个服务提供者，最后进行调用。

#### 负载均衡 ####

抽象类`AbstractLoadBalance`,包含了权重计算，如果没有配置权重则默认时100。权重并不是配置多少就是多少的，获取权重时会从url中获取服务提供者启动时间戳，默认如果启动时间没有超过10分钟，判断没有经过预热进行降权，通过uptime/warmup/weight进行计算，启动时间越长则权重越高，直到达到配置的权重值。如果超过10分钟则直接返回配置的权重值。

##### 权重随机算法的 RandomLoadBalance #####

根据权重分为几个区间，假设权重为2，4，6，那么区间则为0-1，2-5，6-11，通过生成0-11的随机数，命中哪个区间就调用哪个invoke。获取到随机数后，通过遍历invoke集合，减去每个invoke权重值，当随机数剪成负数，说明位于这个区间，返回当前遍历的invoke。缺点是调用次数可能不够平均。

##### 基于最少活跃调用数算法的 LeastActiveLoadBalance #####

最小活跃数指的是活跃数低的机器会优先分配，初始每个服务的活跃数都为0，每收到一个服务请求则活跃数加1，完成一个请求处理则活跃数减去1，即根据当前处理中的请求数决定分配，确保压力平衡。而且支持权重，如果活跃数相同，则权重高的优先分配，权重也想等则随机分配。  
2.6.4有bug，总的权重没把降权算上，减去权重时有算降权，可能导致最终没有一个invoke被选中。还有2.6.4权重值如果设置为1可能会导致不会被选中，临界值问题，也是bug。  
http://dubbo.apache.org/zh-cn/docs/source\_code\_guide/loadbalance.html

##### 基于 hash 一致性的 ConsistentHashLoadBalance #####

基于treemap实现的一致性hash,虚拟节点默认每个invoke为`160`个，分布虚拟节点时.  
分配请求命中的节点时，通过参数进行hash，所以相同的参数会被hash到同一个节点，不关心权重。

内部包含`ConsistentHashSelector`类，基于`TreeMap`实现一致性hash，一个service一个选择器。  
每个invoke进行40次循环，url+i 进行md5再hash,再4次内循环一个值，将invoke放到对应的位置。每个invoke最终会在环上有160个节点。

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);
            }
        }
    }

##### 基于加权轮询算法的 RoundRobinLoadBalance #####

轮询算法指的是每个机器轮流处理，但是会导致性能好的机器和性能差的机器得到相同的请求量。所以允许通过加权来控制。2.6.4以前轮流处理的同时，通过一个mod值记录当前处理到第几次，然后循环mod次机器集合去将权重-1，如果某个机器的权重值为0，则轮到这个机器时会跳过，最后处于哪个机器就返回哪个机器。

这个算法有一个问题是时间复杂度是0(n),每次都要减1，那么如果权重很大，mod值很大的时候，要经过很多次遍历才能拿到最终处理的机器。

2.6.4之后每个服务器对应两个权重，分别为 weight 和 currentWeight。其中 weight 是固定的，currentWeight 会动态调整，初始值为0。当有新的请求进来时，遍历服务器列表，让它的 currentWeight 加上自身权重。遍历完成后，找到最大的 currentWeight，并将其减去权重总和，然后返回相应的服务器即可。  
\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-sRilbFCV-1585712715352)(evernotecid://EA56B0F8-13B5-4B11-8191-1E827D9FC2A6/appyinxiangcom/22897236/ENResource/p20)\]

#### 服务调用过程源码 ####

消费者通过代理对象调用远程服务，将数据进行编码，然后通过netty客户端发送给服务端，服务端解码后将请求派发给指定的线程池，线程池调用具体的服务，然后相应请求，再将结果返回。

1.  `MockClusterInvoker`。服务调用时，会判断是否包含`mock`配置，有的话会根据mock做降级调用，比如直接返回不调用，或者调用加个try catch，在异常时代调用mock的代码。
2.  `AbstractInvoker`。调用时判断是否为`异步调用`，异步调用直接返回，需要返回值则将`Future`设置到`rpcContent`,`同步调用则直接get等待返回值`，此时会用户线程进入超时`await`状态，在服务端响应后会唤醒，在超时后仍没收到响应会报错超时异常。具体返回值获取DefaultFuture类处理,`DefaultFuture`创建时会要求传入requeset对象，包含一个`调用编号`，`DefaultFuture`类有一个`FUTURES`静态map，以调用编号为key，future对象为value，以便`响应结果时可以通过调用编号找到对应的future处理`。
3.  最终发送请求，由netty的`NettyClient`的send方法发送，send方法通过`NettyChannel`的send发送，最后通过channel write写数据出去。在发送数据后，会先判断`sent`参数是否为true,为true则等待write结果，如果发送失败就报错。如果sent为false则直接返回。
4.  ExchangeCodec，消息编码。将消息封装成dubbo规定的消息结构，包含魔数（用于`消息校验，标识是dubbo发出的消息，接收方进行校验）、消息长度记录、调用编号`到消息头等，将发送的dubbo版本号、路径、方法名、参数进行序列化存储到结构中。
5.  接受接受放。DecodeableRpcInvocation::decode,解码:`魔数校验，取出消息头消息长度，长度校验，对消息进行解码（in.readUTF()获取版本号、请求路径、方法名、参数值、调用编号），通过返序列化获取参数值`。最后获取到一个request对象
6.  NettyHandler处理request,之后由nettyServer处理，最后交给线程派发模型处理。
7.  **线程派发模型**：dubbo把处理请求的线程称为io线程，如果业务处理只是内存操作可以在io线程执行，如果是耗时操作，不可以在io线程上执行，会影响请求接受，应该把处理派发到线程池中处理。线程派发器Dispatcher，它的职责是创建具有线程派发能力的ChannelHandler，dubbo支持5种派发模型，默认是all类型。由`AllChannelHandler`处理，建立连接、断开链接、处理请求和响应都封装成`ChannelEventRunnable`给线程池处理。  
    \[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-sSFkSlTZ-1585712715355)(evernotecid://EA56B0F8-13B5-4B11-8191-1E827D9FC2A6/appyinxiangcom/22897236/ENResource/p21)\]
8.  `ChannelEventRunnable`把消息再交给各handler处理，比如消息解码，最后是`HeaderExchangeHandler`,判断是双向通信会调用具体的请求方法后，通过channel将响应结果封装到Response send给客户端，出现异常也同样将异常信息封装后返回。同样需要将响应数据进行编码，需要在消息头设置`魔数、响应状态、消息长度、序列化请求结果，调用编号`.
9.  服务消费方接受响应结果。消息解码，根据`调用编号`在DefaultFuture的静态map参数`FUTURES`找到对应的`DefaultFuture`对象，然后将respose设置进去，通过`Condition::signal`唤醒线程。

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-YvpnCGvp-1585712715355)(evernotecid://EA56B0F8-13B5-4B11-8191-1E827D9FC2A6/appyinxiangcom/22897236/ENResource/p48)\]

#### tcp拆包、粘包解决 ####

在TCP网络传输工程中，由于TCP包的缓存大小限制，每次请求数据有可能不在一个TCP包里面，或者也可能多个请求的数据在一个TCP包里面。那么如果合理的decode接受的TCP数据很重要，需要考虑TCP拆包和粘包的问题

dubbo具有高度自定义的协议，消息被分为消息头和消息体。因为高度自定义不能直接用netty进行消息编码和解码。消息头规定为16字节。源码在`ExchangeCodec`。

1.  服务端先创建消息头大小的buff，从buffer(用户缓冲区)中读取，即先把消息头读出来。当读取的消息可读数据还不够16字节时，即还不够消息头的大小，则从channel中读取到buffer,再等待下次tcp包过来继续读取。且会通过消息头的0号位置`魔数`校验。（读取到的消息会暂存到成员变量`NettyCodecAdapter::InternalDecoder::buff`，等到处理完一个完整的dubbo数据包后再置空，如果处理完后，buff还有剩余数据，则会重复流程去读取更多数据）
2.  读到完整的消息头后，取出消息头标志的消息长度，根据消息长度读取消息。同样不够读取会重新去读。
3.  当buffer的数据包含一个完整的dubbo包数据后，开始解码。

发生拆包时，当前tcp没有包含完整数据，会去继续读取放到buffer。直到读取到完整数据包  
发生黏包时，读取完一个完整数据包后，会发现buffer还可读，继续下次循环去读取，接下来可能又出现消息不够，继续等待下次tcp包数据读取的情况。

如此看来tcp包的有序性就很重要了，如果tcp没有保证有序性，那么数据就是乱的了。  
https://blog.csdn.net/zhengyangzkr/article/details/71486851

#### 服务订阅 推和拉 ####

dubbo订阅服务基于zk的是用watch机制，建立长链接，要有心跳检测机制。长链接会占用长期系统资源。  
拉就是主动定时拉取服务列表。

#### dubbo异常处理 ####

ExceptionFilter，服务端将在异常时将异常封装返回，客户端获取结果判断时异常抛出异常。  
自定义异常参考应用篇抛出。

#### 总结 ####

*  服务提供端在项目启动时向注册中心注册服务，在`/分组名/服务权限定名/providers`目录下，创建临时节点，包含ip地址等，同个服务多个机器注册则构成集群，然后以服务地址为key，Exporter为value,Exporter包含服务调用的invoke。消费端请求过来后会通过key找到对应的`invoke`进行调用处理,通过反射调用方法。
 *  服务消费端有一个`服务目录`，会根据注册中心的服务情况动态变成invoke集合，消费端发起调用时会从服务目录获取到invoke集合，然后经过`条件路由`过滤掉一些invoke之后，如果有多个invoke，则构造一个`cluster`对象，再封装返回一个invoke，发起调用时，再通过`cluster`的类型做不同的调用方式。如果是一次只调用一个机器，会通过负载均衡策略选择一个机器的invoker进行调用，而且基于`粘滞`策略，如果不出现问题，同个消费者会一直调用同一个机器的invoke，避免重复创建连接。
 *  获取到一个invoker之后，消费端将消息进行编码，发起调用时会根据`服务降级`策略和`异步同步`策略进行调用，调用时会封装`DefaultFuture`对象，生成`调用编号`,通过`nettyClient`与服务端建立连接，将调用编号请求信息等通过`nettyChannel`发送给服务端，然后调用线程进入await有限等待。服务端接收到消息后，对消息进行`解码`，将消息交给对应的方法处理后将处理结果封装成respose后，将调用编号和处理结果通过`nettyChannel`发送给消费端，消费端通过`调用编号`找到对应的`DefaultFuture`唤醒线程得到请求结果。

服务端和消费端都会通过nettyChannel去启动一个server,创建invoke，客户端请求会把请求地址、端口、请求id也带在消息体中，dubbo处理完消息后，再通过请求地址发送消费端，消费端根据id找到请求线程，将请求结果设置到id对应的线程后，唤醒请求线程。