Dubbo源码解析-高可用集群
Posted 黑马程序员官方
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Dubbo源码解析-高可用集群相关的知识，希望对你有一定的参考价值。
dubbo源码解析-高可用集群
服务集群的概述
概述
为了避免单点故障，现在的应用通常至少会部署在两台服务器上，这样就组成了集群。集群就是单机的多实例，在多个服务器上部署多个服务，每个服务就是一个节点，部署N个节点，处理业务的能力就提升 N倍（大约），这些节点的集合就叫做集群。
管理控制台
目前的管理控制台已经发布0.1版本，结构上采取了前后端分离的方式，前端使用Vue和Vuetify分别作为javascript框架和UI框架，后端采用Spring Boot框架。既可以按照标准的Maven方式进行打包，部署，也可以采用前后端分离的部署方式，方便开发，功能上，目前具备了服务查询，服务治理(包括Dubbo2.7中新增的治理规则)以及服务测试三部分内容。
Maven方式部署
- 安装
git clone https://github.com/apache/dubbo-admin.git
cd dubbo-admin
mvn clean package
cd dubbo-admin-distribution/target
java -jar dubbo-admin-0.1.jar
前后端分离部署
- 前端
cd dubbo-admin-ui
npm install
npm run dev
- 后端
cd dubbo-admin-server
mvn clean package
cd target
java -jar dubbo-admin-server-0.1.jar
- 访问
http://localhost:8081 - 前后端分离模式下，前端的修改可以实时生效
环境搭建
略
集群调用存在的问题
- 负载均衡
- 集群容错
- 服务治理
集群的调用过程
调用过程
在对集群相关代码进行分析之前，这里有必要先来介绍一下集群容错的所有组件。包含 Cluster、Cluster Invoker、Directory、Router 和 LoadBalance 等。
集群工作过程可分为两个阶段，第一个阶段是在服务消费者初始化期间，集群 Cluster 实现类为服务消费者创建 Cluster Invoker 实例，即上图中的 merge 操作。第二个阶段是在服务消费者进行远程调用时。以 FailoverClusterInvoker 为例，该类型 Cluster Invoker 首先会调用 Directory 的 list 方法列举 Invoker 列表（可将 Invoker 简单理解为服务提供者）。Directory 的用途是保存 Invoker，可简单类比为 List。其实现类 RegistryDirectory 是一个动态服务目录，可感知注册中心配置的变化，它所持有的 Invoker 列表会随着注册中心内容的变化而变化。每次变化后，RegistryDirectory 会动态增删 Invoker，并调用 Router 的 route 方法进行路由，过滤掉不符合路由规则的 Invoker。当 FailoverClusterInvoker 拿到 Directory 返回的 Invoker 列表后，它会通过 LoadBalance 从 Invoker 列表中选择一个 Invoker。最后 FailoverClusterInvoker 会将参数传给 LoadBalance 选择出的 Invoker 实例的 invoke 方法，进行真正的远程调用。
组件介绍
- Directory：它代表多个Invoker，从methodInvokerMap提取，但是他的值是动态，例如注册中心的变更。
- Router：负责从多个Invoker中按路由规则选出子集，例如应用隔离或读写分离或灰度发布等等
- Cluster：将Directory中的多个Invoker伪装成一个Invoker,来容错，调用失败重试。
- LoadBalance：从多个Invoker选取一个做本次调用，具体包含很多种负载均衡算法。
- Invoker：Provider中的一个可调用接口。例如DemoService
集群容错
在分布式系统中，集群某个某些节点出现问题是大概率事件，因此在设计分布式RPC框架的过程中，必须要把失败作为设计的一等公民来对待。一次调用失败之后，应该如何选择对失败的选择策略，这是一个见仁见智的问题，每种策略可能都有自己独特的应用场景。因此，作为框架来说，应当针对不同场景提供多种策略，供用户进行选择。
在Dubbo设计中，通过Cluster这个接口的抽象，把一组可供调用的Provider信息组合成为一个统一的Invoker供调用方进行调用。经过路由规则过滤，负载均衡选址后，选中一个具体地址进行调用，如果调用失败，则会按照集群配置的容错策略进行容错处理。
内置集群容错方式
Dubbo默认内置了若干容错策略，如果不能满足用户需求，则可以通过自定义容错策略进行配置
Dubbo主要内置了如下几种策略：
- Failover(失败自动切换)
- Failsafe(失败安全)
- Failfast(快速失败)
- Failback(失败自动恢复)
- Forking(并行调用)
- Broadcast(广播调用)
这些名称比较相似，概念也比较容易混淆，下面逐一进行解释。
Failover(失败自动切换)
Failover是高可用系统中的一个常用概念，服务器通常拥有主备两套机器配置，如果主服务器出现故障，则自动切换到备服务器中，从而保证了整体的高可用性。
Dubbo也借鉴了这个思想，并且把它作为Dubbo默认的容错策略。当调用出现失败的时候，根据配置的重试次数，会自动从其他可用地址中重新选择一个可用的地址进行调用，直到调用成功，或者是达到重试的上限位置。
Dubbo里默认配置的重试次数是2，也就是说，算上第一次调用，最多会调用3次。
其配置方法，容错策略既可以在服务提供方配置，也可以服务调用方进行配置。而重试次数的配置则更为灵活，既可以在服务级别进行配置，也可以在方法级别进行配置。具体优先顺序为：
服务调用方方法级配置 > 服务调用方服务级配置 > 服务提供方方法级配置 > 服务提供方服务级配置
以XML方式为例，具体配置方法如下：
服务提供方，服务级配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="failover" retries="2" />
服务提供方，方法级配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService"cluster="failover">
<dubbo:method name="sayHello" retries="2" />
</dubbo:reference>
服务调用方，服务级配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="failover" retries="1"/>
服务调用方，方法级配置：
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="failover">
<dubbo:method name="sayHello" retries="3" />
</dubbo:reference>
Failover可以自动对失败进行重试，对调用者屏蔽了失败的细节，但是Failover策略也会带来一些副作用：
- 重试会额外增加一下开销，例如增加资源的使用，在高负载系统下，额外的重试可能让系统雪上加霜。
- 重试会增加调用的响应时间。
- 某些情况下，重试甚至会造成资源的浪费。考虑一个调用场景，A->B->C，如果A处设置了超时100ms，再B->C的第一次调用完成时已经超过了100ms，但很不幸B->C失败，这时候会进行重试，但其实这时候重试已经没有意义，因此在A看来这次调用已经超时，A可能已经开始执行其他逻辑。
Failsafe(失败安全)
失败安全策略的核心是即使失败了也不会影响整个调用流程。通常情况下用于旁路系统或流程中，它的失败不影响核心业务的正确性。在实现上，当出现调用失败时，会忽略此错误，并记录一条日志，同时返回一个空结果，在上游看来调用是成功的。
应用场景，可以用于写入审计日志等操作。
具体配置方法：
服务提供方，服务级配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="failsafe" />
服务调用方，服务级配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="failsafe"/>
其中服务调用方配置优先于服务提供方配置。
Failfast(快速失败)
某些业务场景中，某些操作可能是非幂等的，如果重复发起调用，可能会导致出现脏数据等。例如调用某个服务，其中包含一个数据库的写操作，如果写操作完成，但是在发送结果给调用方的过程中出错了，那么在调用发看来这次调用失败了，但其实数据写入已经完成。这种情况下，重试可能并不是一个好策略，这时候就需要使用到Failfast策略，调用失败立即报错。让调用方来决定下一步的操作并保证业务的幂等性。
具体配置方法：
服务提供方，服务级配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="failfast" />
服务调用方，服务级配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="failfast"/>
其中服务调用方配置优先于服务提供方配置。
Failback(失败自动恢复)
Failback通常和Failover两个概念联系在一起。在高可用系统中，当主机发生故障，通过Failover进行主备切换后，待故障恢复后，系统应该具备自动恢复原始配置的能力。
Dubbo中的Failback策略中，如果调用失败，则此次失败相当于Failsafe，将返回一个空结果。而与Failsafe不同的是，Failback策略会将这次调用加入内存中的失败列表中，对于这个列表中的失败调用，会在另一个线程中进行异步重试，重试如果再发生失败，则会忽略，即使重试调用成功，原来的调用方也感知不到了。因此它通常适合于，对于实时性要求不高，且不需要返回值的一些异步操作。
具体配置方法：
服务提供方，服务级配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="failsafe" />
服务调用方，服务级配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="failsafe"/>
其中服务调用方配置优先于服务提供方配置。
按照目前的实现，Failback策略还有一些局限，例如内存中的失败调用列表没有上限，可能导致堆积，异步重试的执行间隔无法调整，默认是5秒。
Forking(并行调用)
上述几种策略中，主要都是针对调用失败发生后如何进行弥补的角度去考虑的，而Forking策略则跟上述几种策略不同，是一种典型的用成本换时间的思路。即第一次调用的时候就同时发起多个调用，只要其中一个调用成功，就认为成功。在资源充足，且对于失败的容忍度较低的场景下，可以采用此策略。
具体配置方法：
服务提供方，服务级配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="forking" />
服务调用方，服务级配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="forking"/>
其中服务调用方配置优先于服务提供方配置。
Broadcast(广播调用)
在某些场景下，可能需要对服务的所有提供者进行操作，此时可以使用广播调用策略。此策略会逐个调用所有提供者，只要任意有一个提供者出错，则认为此次调用出错。通常用于通知所有提供者更新缓存或日志等本地资源信息。
具体配置方法：
服务提供方，服务级配置
<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" cluster="broadcast" />
服务调用方，服务级配置
<dubbo:reference id="demoService" interface="org.apache.dubbo.demo.DemoService" cluster="broadcast"/>
其中服务调用方配置优先于服务提供方配置。
集群容错调优
下表对各种策略做一个简单对比，
| 策略名称 | 优点 | 缺点 |
|---|---|---|
| Failover | 对调用者屏蔽调用失败的信息 | 增加RT，额外资源开销，资源浪费 |
| Failfast | 业务快速感知失败状态进行自主决策 | 产生较多报错的信息 |
| Failsafe | 即使失败了也不会影响核心流程 | 对于失败的信息不敏感，需要额外的监控 |
| Failback | 失败自动异步重试 | 重试任务可能堆积 |
| Forking | 并行发起多个调用，降低失败概率 | 消耗额外的机器资源，需要确保操作幂等性 |
| Broadcast | 支持对所有的服务提供者进行操作 | 资源消耗很大 |
综上我们得知，不同的容错策略往往对应不同的业务处理，这里做一个总结如下：
- **Failover **：通常用于对调用rt不敏感的场景，如读操作；但重试会带来更长延迟
- Failfast ：通常用于非幂等性操作，需要快速感知失败的场景；比如新增记录
- **Failsafe **：通常用于旁路系统，失败不影响核心流程正确性的场景；如日志记录
- Failback ：通常用于对于实时性要求不高，且不需要返回值的一些异步操作的场景
- Forking ：通常用于资源充足，且对于失败的容忍度较低，实时性要求高的读操作，但需要浪费更多服务资源
- Broadcast：如通知所有提供者更新缓存或日志等本地资源信息
源码分析
我们在上一章看到了两个概念，分别是集群接口 Cluster 和 Cluster Invoker，这两者是不同的。Cluster 是接口，而 Cluster Invoker 是一种 Invoker。服务提供者的选择逻辑，以及远程调用失败后的的处理逻辑均是封装在 Cluster Invoker 中。那么 Cluster 接口和相关实现类有什么用呢？用途比较简单，仅用于生成 Cluster Invoker。下面我们来看一下源码。
public class FailoverCluster implements Cluster
public final static String NAME = "failover";
@Override
public <T> Invoker<T> join(Directory<T> directory) throws RpcException
// 创建并返回 FailoverClusterInvoker 对象
return new FailoverClusterInvoker<T>(directory);
如上，FailoverCluster 总共就包含这几行代码，用于创建 FailoverClusterInvoker 对象，很简单。下面再看一个。
public class FailbackCluster implements Cluster
public final static String NAME = "failback";
@Override
public <T> Invoker<T> join(Directory<T> directory) throws RpcException
// 创建并返回 FailbackClusterInvoker 对象
return new FailbackClusterInvoker<T>(directory);
如上，FailbackCluster 的逻辑也是很简单，无需解释了。所以接下来，我们把重点放在各种 Cluster Invoker 上
Cluster Invoker
我们首先从各种 Cluster Invoker 的父类 AbstractClusterInvoker 源码开始说起。前面说过，集群工作过程可分为两个阶段，第一个阶段是在服务消费者初始化期间，这个在服务引用那篇文章中分析过，就不赘述。第二个阶段是在服务消费者进行远程调用时，此时 AbstractClusterInvoker 的 invoke 方法会被调用。列举 Invoker，负载均衡等操作均会在此阶段被执行。因此下面先来看一下 invoke 方法的逻辑。
public Result invoke(final Invocation invocation) throws RpcException
checkWhetherDestroyed();
LoadBalance loadbalance = null;
// 绑定 attachments 到 invocation 中.
Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
if (contextAttachments != null && contextAttachments.size() != 0)
((RpcInvocation) invocation).addAttachments(contextAttachments);
// 列举 Invoker
List<Invoker<T>> invokers = list(invocation);
if (invokers != null && !invokers.isEmpty())
// 加载 LoadBalance
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
.getMethodParameter(RpcUtils.getMethodName(invocation), Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
// 调用 doInvoke 进行后续操作
return doInvoke(invocation, invokers, loadbalance);
// 抽象方法，由子类实现
protected abstract Result doInvoke(Invocation invocation, List<Invoker<T>> invokers,
LoadBalance loadbalance) throws RpcException;
AbstractClusterInvoker 的 invoke 方法主要用于列举 Invoker，以及加载 LoadBalance。最后再调用模板方法 doInvoke 进行后续操作。下面我们来看一下 Invoker 列举方法 list(Invocation) 的逻辑，如下：
protected List<Invoker<T>> list(Invocation invocation) throws RpcException
// 调用 Directory 的 list 方法列举 Invoker
List<Invoker<T>> invokers = directory.list(invocation);
return invokers;
如上，AbstractClusterInvoker 中的 list 方法做的事情很简单，只是简单的调用了 Directory 的 list 方法，没有其他更多的逻辑了。Directory 即相关实现类在前文已经分析过，这里就不多说了。接下来，我们把目光转移到 AbstractClusterInvoker 的各种实现类上，来看一下这些实现类是如何实现 doInvoke 方法逻辑的。
FailoverClusterInvoker
FailoverClusterInvoker 在调用失败时，会自动切换 Invoker 进行重试。默认配置下，Dubbo 会使用这个类作为缺省 Cluster Invoker。下面来看一下该类的逻辑。
public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T>
// 省略部分代码
@Override
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException
List<Invoker<T>> copyinvokers = invokers;
checkInvokers(copyinvokers, invocation);
// 获取重试次数
int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
if (len <= 0)
len = 1;
RpcException le = null;
List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size());
Set<String> providers = new HashSet<String>(len);
// 循环调用，失败重试
for (int i = 0; i < len; i++)
if (i > 0)
checkWhetherDestroyed();
// 在进行重试前重新列举 Invoker，这样做的好处是，如果某个服务挂了，
// 通过调用 list 可得到最新可用的 Invoker 列表
copyinvokers = list(invocation);
// 对 copyinvokers 进行判空检查
checkInvokers(copyinvokers, invocation);
// 通过负载均衡选择 Invoker
Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
// 添加到 invoker 到 invoked 列表中
invoked.add(invoker);
// 设置 invoked 到 RPC 上下文中
RpcContext.getContext().setInvokers((List) invoked);
try
// 调用目标 Invoker 的 invoke 方法
Result result = invoker.invoke(invocation);
return result;
catch (RpcException e)
if (e.isBiz())
throw e;
le = e;
catch (Throwable e)
le = new RpcException(e.getMessage(), e);
finally
providers.add(invoker.getUrl().getAddress());
// 若重试失败，则抛出异常
throw new RpcException(..., "Failed to invoke the method ...");
如上，FailoverClusterInvoker 的 doInvoke 方法首先是获取重试次数，然后根据重试次数进行循环调用，失败后进行重试。在 for 循环内，首先是通过负载均衡组件选择一个 Invoker，然后再通过这个 Invoker 的 invoke 方法进行远程调用。如果失败了，记录下异常，并进行重试。重试时会再次调用父类的 list 方法列举 Invoker。整个流程大致如此，不是很难理解。下面我们看一下 select 方法的逻辑。
protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException
if (invokers == null || invokers.isEmpty())
return null;
// 获取调用方法名
String methodName = invocation == null ? "" : invocation.getMethodName();
// 获取 sticky 配置，sticky 表示粘滞连接。所谓粘滞连接是指让服务消费者尽可能的
// 调用同一个服务提供者，除非该提供者挂了再进行切换
boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName, Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY);
// 检测 invokers 列表是否包含 stickyInvoker，如果不包含，
// 说明 stickyInvoker 代表的服务提供者挂了，此时需要将其置空
if (stickyInvoker != null && !invokers.contains(stickyInvoker))
stickyInvoker = null;
// 在 sticky 为 true，且 stickyInvoker != null 的情况下。如果 selected 包含
// stickyInvoker，表明 stickyInvoker 对应的服务提供者可能因网络原因未能成功提供服以上是关于Dubbo源码解析-高可用集群的主要内容，如果未能解决你的问题，请参考以下文章
Java集群优化——dubbo+zookeeper构建高可用分布式集群
[置顶] Java集群优化——dubbo+zookeeper构建高可用分布式集群