Dubbo源码(七) - 集群

前言

本文基于Dubbo2.6.x版本，中文注释版源码已上传github：xiaoguyu/dubbo
集群(cluster)就是一组计算机，它们作为一个总体向用户提供一组网络资源。这些单个的计算机系统就是集群的节点(node)。
在Dubbo中，为了避免单点故障，同一个服务允许有多个服务提供者，也允许同时连接多个注册中心。那么，服务消费者引用服务时，该请求哪个注册中心的服务提供者以及调用失败之后该如何处理呢？这些就是Dubbo集群所做的事。
集群容错

在分析集群源码之前，先看看集群容错的所有组件，下图是官方文档的组件图

Dubbo 定义了集群接口 Cluster 以及 Cluster Invoker：

• Cluster 是接口，其只有一个方法，负责生成Cluster Invoker
  
• Cluster Invoker继承了Invoker接口，是一个 Invoker，是主要逻辑实现的地方
  

将上图从中间切分，可将集群工作过程分为两个阶段，左边为第一阶段。

• 第一个阶段是在服务消费者初始化期间。
  集群 Cluster 实现类为服务消费者创建 Cluster Invoker，即图上的 merge 操作，也就是将多个服务提供者合并为一个 Cluster Invoker
  
• 第二个阶段是在服务消费者进行远程调用时。
  步骤大体上就如图所示：list → route → select → invoke
  
  
  
  • list：从服务目录拿到 invoker 集合
    
  • route：通过路由过滤出符合规则的 invoker 集合
    
  • select：通过负载均衡从 invoker 集合中选择一个
    
  • invoke：执行 invoker 的 invoke 方法，进行真正的远程调用
    
  其中，list、route操作在之前文章讲过了，传送门：《服务目录》、《服务路由》
  select 不是本文重点，后续负载均衡时讲解。
  
  

以上就是集群工作的整个流程，这里并没有介绍集群是如何容错的，也就是 invoke 步骤调用失败的处理。Dubbo提供了多种容错方式：集群容错示例
下面的源码我们以默认的 Failover Cluster - 失败自动切换 进行分析
源码分析

Cluster

首先来看看 Cluster 接口，这是一个自适应拓展类，默认实现为FailoverCluster

1. public class FailoverCluster implements Cluster {
3.     public final static String NAME = "failover";
5.     @Override
6.     public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
7.         return new FailoverClusterInvoker<T>(directory);
8.     }
10. }

复制代码

前面讲了，Cluster 的作用就是将多个服务提供者合并为一个 Cluster Invoker
多个服务提供者合并也就是 服务目录(Directory) 中的 invoker 集合。join 方法返回了一个 Cluster Invoker
接下来，我们看看调用路径。 Cluster 接口在多个地方被调用，我们看服务消费者初始化期间的调用。

1. // 调用路径如下：
2. // ReferenceBean#getObject()
3. // ReferenceConfig#get()
4. // ReferenceConfig#init()
5. // ReferenceConfig#createProxy(Map<String, String> map)
6. // RegistryProtocol#refer(Class<T> type, URL url)
7. // RegistryProtocol#doRefer(Cluster cluster, Registry registry, Class<T> type, URL url)
8. private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
9.     // 创建 RegistryDirectory 实例
10.     RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
11.     // 设置注册中心和协议
12.     directory.setRegistry(registry);
13.     directory.setProtocol(protocol);
14.     // all attributes of REFER_KEY
15.     Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
16.     // 生成服务消费者链接
17.     URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, parameters.remove(Constants.REGISTER_IP_KEY), 0, type.getName(), parameters);
18.     // 注册服务消费者，在 consumers 目录下新节点
19.     if (!Constants.ANY_VALUE.equals(url.getServiceInterface())
20.             && url.getParameter(Constants.REGISTER_KEY, true)) {
21.         URL registeredConsumerUrl = getRegisteredConsumerUrl(subscribeUrl, url);
22.         registry.register(registeredConsumerUrl);
23.         directory.setRegisteredConsumerUrl(registeredConsumerUrl);
24.     }
26.     // 订阅 providers、configurators、routers 等节点数据
27.     directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
28.             Constants.PROVIDERS_CATEGORY
29.                     + "," + Constants.CONFIGURATORS_CATEGORY
30.                     + "," + Constants.ROUTERS_CATEGORY));
32.     // 一个注册中心可能有多个服务提供者，因此这里需要将多个服务提供者合并为一个
33.     Invoker invoker = cluster.join(directory);
34.     ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
35.     return invoker;
36. }

复制代码

调用在倒数第三行代码。
如果看过我之前写的《服务引用》那篇文章，想必对 doRefer 方法不陌生了。在服务目录订阅完注册中心的数据后，就调用 join 方法生成 Cluster Invoker
啰嗦多一句：
可以这么理解，实际负责远程调用的，是服务目录中的 invoker 集合中的 invoker，而 Cluster Invoker 则对服务目录中的 invoker 集合进行处理。
Cluster Invoker

默认的 Cluster Invoker 是FailoverClusterInvoker，既然是一个 Invoker，我们就从它的 invoke 方法入手。
AbstractClusterInvoker

invoke 方法在它的父类AbstractClusterInvoker中

1. public Result invoke(final Invocation invocation) throws RpcException {
2.     checkWhetherDestroyed();
3.     LoadBalance loadbalance = null;
5.     // 绑定 attachments 到 invocation 中.
6.     Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
7.     if (contextAttachments != null && contextAttachments.size() != 0) {
8.         ((RpcInvocation) invocation).addAttachments(contextAttachments);
9.     }
11.     // 列举 Invoker
12.     List<Invoker<T>> invokers = list(invocation);
13.     if (invokers != null && !invokers.isEmpty()) {
14.         // 加载 LoadBalance
15.         loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
16.                 .getMethodParameter(RpcUtils.getMethodName(invocation), Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
17.     }
18.     RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
19.     // 调用 doInvoke 进行后续操作
20.     return doInvoke(invocation, invokers, loadbalance);
21. }

复制代码

invoke 方法逻辑也很简单：

• 列举 invoker
  
• 加载 LoadBalance（自适应拓展类）
  
• 调用 doInvoke 进行后续操作
  

其中列举 invoker 如下

1. protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
2.     List<Invoker<T>> invokers = directory.list(invocation);
3.     return invokers;
4. }

复制代码

list 方法就是调用服务目录的 list 方法，里面做了两件事（结合前面的组件图）：

• list：从服务目录拿到 invoker 集合
  
• route：通过路由过滤出符合规则的 invoker 集合
  

FailoverClusterInvoker

doInvoke 方法具体实现在FailoverClusterInvoker中，此 invoker 的容错方式为失败自动切换。

1. public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
2.     List<Invoker<T>> copyinvokers = invokers;
3.     checkInvokers(copyinvokers, invocation);
4.     String methodName = RpcUtils.getMethodName(invocation);
5.     // 获取重试次数
6.     int len = getUrl().getMethodParameter(methodName, Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
7.     if (len <= 0) {
8.         len = 1;
9.     }
10.     // 循环调用，失败重试
11.     RpcException le = null; // last exception.
12.     List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size()); // invoked invokers.
13.     Set<String> providers = new HashSet<String>(len);
14.     for (int i = 0; i < len; i++) {
15.         if (i > 0) {
16.             checkWhetherDestroyed();
17.             // 在进行重试前重新列举 Invoker，这样做的好处是，如果某个服务挂了，
18.             // 通过调用 list 可得到最新可用的 Invoker 列表
19.             copyinvokers = list(invocation);
20.             // check again
21.             // 对 copyinvokers 进行判空检查
22.             checkInvokers(copyinvokers, invocation);
23.         }
24.         // 通过负载均衡选择 Invoker
25.         Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
26.         // 添加到 invoker 到 invoked 列表中
27.         invoked.add(invoker);
28.         // 设置 invoked 到 RPC 上下文中
29.         RpcContext.getContext().setInvokers((List) invoked);
30.         try {
31.             // 调用目标 Invoker 的 invoke 方法
32.             Result result = invoker.invoke(invocation);
33.             return result;
34.         } catch (RpcException e) {
35.             if (e.isBiz()) { // biz exception.
36.                 throw e;
37.             }
38.             le = e;
39.         } catch (Throwable e) {
40.             le = new RpcException(e.getMessage(), e);
41.         } finally {
42.             providers.add(invoker.getUrl().getAddress());
43.         }
44.     }
45.     // 若重试失败，则抛出异常
46.     throw new RpcException(xxx);
47. }

复制代码

doInvoke 方法代码量不少，但是逻辑简化之后也很简单，就是根据重试次数，在 for 循环中进行远程调用，成功则返回，失败就重试。如果重试次数耗尽还无法调用成功，则抛出异常。
从这里可以知道，Dubbo的默认失败重试次数是3次。
此方法中我们关注下 select 方法，它负责从 invoker 集合中选出一个 infoker

1. protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
2.     if (invokers == null || invokers.isEmpty())
3.         return null;
4.     // 获取调用方法名
5.     String methodName = invocation == null ? "" : invocation.getMethodName();
7.     // 获取 sticky 配置，sticky 表示粘滞连接。所谓粘滞连接是指让服务消费者尽可能的
8.     // 调用同一个服务提供者，除非该提供者挂了再进行切换
9.     boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName, Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY);
10.     {
11.         // 检测 invokers 列表是否包含 stickyInvoker，如果不包含，
12.         // 说明 stickyInvoker 代表的服务提供者挂了，此时需要将其置空
13.         if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
14.             stickyInvoker = null;
15.         }
16.         // 在 sticky 为 true，且 stickyInvoker != null 的情况下。如果 selected 包含
17.         // stickyInvoker，表明 stickyInvoker 对应的服务提供者可能因网络原因未能成功提供服务。
18.         // 但是该提供者并没挂，此时 invokers 列表中仍存在该服务提供者对应的 Invoker。
19.         if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
20.             // availablecheck 表示是否开启了可用性检查，如果开启了，则调用 stickyInvoker 的
21.             // isAvailable 方法进行检查，如果检查通过，则直接返回 stickyInvoker。
22.             if (availablecheck && stickyInvoker.isAvailable()) {
23.                 return stickyInvoker;
24.             }
25.         }
26.     }
28.     // 如果线程走到当前代码处，说明前面的 stickyInvoker 为空，或者不可用。
29.     // 此时继续调用 doSelect 选择 Invoker
30.     Invoker<T> invoker = doSelect(loadbalance, invocation, invokers, selected);
32.     // 如果 sticky 为 true，则将负载均衡组件选出的 Invoker 赋值给 stickyInvoker
33.     if (sticky) {
34.         stickyInvoker = invoker;
35.     }
36.     return invoker;
37. }

复制代码

select 方法主要处理对粘滞连接特性的支持。注释写的很清楚了。选择 invoker 的操作在 doSelect 方法

1. private Invoker<T> doSelect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
2.     if (invokers == null || invokers.isEmpty())
3.         return null;
4.     if (invokers.size() == 1)
5.         return invokers.get(0);
6.     if (loadbalance == null) {
7.         // 如果 loadbalance 为空，这里通过 SPI 加载 Loadbalance，默认为 RandomLoadBalance
8.         loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
9.     }
10.     // 通过负载均衡组件选择 Invoker
11.     Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);
13.     // 如果 selected 包含负载均衡选择出的 Invoker，或者该 Invoker 无法经过可用性检查，此时进行重选
14.     if ((selected != null && selected.contains(invoker))
15.             || (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
16.         try {
17.             // 进行重选
18.             Invoker<T> rinvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
19.             if (rinvoker != null) {
20.                 invoker = rinvoker;
21.             } else {
22.                 // rinvoker 为空，定位 invoker 在 invokers 中的位置
23.                 int index = invokers.indexOf(invoker);
24.                 try {
25.                     // 获取 index + 1 位置处的 Invoker，以下代码等价于：
26.                     //     invoker = invokers.get((index + 1) % invokers.size());
27.                     invoker = index < invokers.size() - 1 ? invokers.get(index + 1) : invokers.get(0);
28.                 } catch (Exception e) {
29.                     logger.warn(e.getMessage() + " may because invokers list dynamic change, ignore.", e);
30.                 }
31.             }
32.         } catch (Throwable t) {
33.             logger.error("cluster reselect fail reason is :" + t.getMessage() + " if can not solve, you can set cluster.availablecheck=false in url", t);
34.         }
35.     }
36.     return invoker;
37. }

复制代码

这里通过负载均衡选出 invoker，如果 invoker 在 selected 中（就是在doInvoke方法中调用失败的invoker）或者不可用，则调用 reselect 方法进行重选。如果重选还是选不出 invoker，则返回 invoker 集合中的下一个元素。这里的繁琐判断，就是为了尽量保证拿到可用的 invoker
我们继续看看 reselect 方法

1. private Invoker<T> reselect(LoadBalance loadbalance, Invocation invocation,
2.                                 List<Invoker<T>> invokers, List<Invoker<T>> selected, boolean availablecheck)
3.             throws RpcException {
5.     List<Invoker<T>> reselectInvokers = new ArrayList<Invoker<T>>(invokers.size() > 1 ? (invokers.size() - 1) : invokers.size());
7.     // 下面的 if-else 分支逻辑有些冗余，pull request #2826 对这段代码进行了简化，可以参考一下
8.     // 根据 availablecheck 进行不同的处理
9.     if (availablecheck) { // invoker.isAvailable() should be checked
10.         for (Invoker<T> invoker : invokers) {
11.             if (invoker.isAvailable()) {
12.                 if (selected == null || !selected.contains(invoker)) {
13.                     reselectInvokers.add(invoker);
14.                 }
15.             }
16.         }
17.         if (!reselectInvokers.isEmpty()) {
18.             return loadbalance.select(reselectInvokers, getUrl(), invocation);
19.         }
20.     } else { // do not check invoker.isAvailable()
21.         for (Invoker<T> invoker : invokers) {
22.             if (selected == null || !selected.contains(invoker)) {
23.                 reselectInvokers.add(invoker);
24.             }
25.         }
26.         if (!reselectInvokers.isEmpty()) {
27.             return loadbalance.select(reselectInvokers, getUrl(), invocation);
28.         }
29.     }
30.     {
31.         // 若线程走到此处，说明 reselectInvokers 集合为空，此时不会调用负载均衡组件进行筛选。
32.         // 这里从 selected 列表中查找可用的 Invoker，并将其添加到 reselectInvokers 集合中
33.         if (selected != null) {
34.             for (Invoker<T> invoker : selected) {
35.                 if ((invoker.isAvailable()) // available first
36.                         && !reselectInvokers.contains(invoker)) {
37.                     reselectInvokers.add(invoker);
38.                 }
39.             }
40.         }
41.         if (!reselectInvokers.isEmpty()) {
42.             return loadbalance.select(reselectInvokers, getUrl(), invocation);
43.         }
44.     }
45.     return null;
46. }

复制代码

这个方法可以分成两部分：

• 在非 selected 的 invoker 集合中，调用负载均衡选择一个 invoker
  
• 在步骤1无法选出 invoker 时，在 selected 中选出 invoker
  

至此，Dubbo的集群就讲完了。负载均衡有空再说。
再论Cluster

前面我们提到 Cluster 接口在多个地方被调用，也讲了同一个服务有多个服务提供者时的处理。那么，有多个注册中心呢，该如何处理？

1. // 类ReferenceConfig
2. private T createProxy(Map<String, String> map) {
3.     ......
5.     // 本地引用
6.     if (isJvmRefer) {
7.         ......
8.     // 远程引用
9.     } else {
10.         ......
12.         // 单个注册中心或服务提供者(服务直连，下同)
13.         if (urls.size() == 1) {
14.             // 调用 RegistryProtocol 的 refer 构建 Invoker 实例
15.             invoker = refprotocol.refer(interfaceClass, urls.get(0));
17.         // 多个注册中心或多个服务提供者，或者两者混合
18.         } else {
19.             List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
20.             URL registryURL = null;
22.             // 获取所有的 Invoker
23.             for (URL url : urls) {
24.                 // 通过 refprotocol 调用 refer 构建 Invoker，refprotocol 会在运行时
25.                 // 根据 url 协议头加载指定的 Protocol 实例，并调用实例的 refer 方法
26.                 invokers.add(refprotocol.refer(interfaceClass, url));
27.                 if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
28.                     registryURL = url; // use last registry url
29.                 }
30.             }
31.             if (registryURL != null) { // registry url is available
32.                 // use AvailableCluster only when register's cluster is available
33.                 // 如果注册中心链接不为空，则将使用 AvailableCluster
34.                 URL u = registryURL.addParameterIfAbsent(Constants.CLUSTER_KEY, AvailableCluster.NAME);
35.                 // 创建 StaticDirectory 实例，并由 Cluster 对多个 Invoker 进行合并
36.                 invoker = cluster.join(new StaticDirectory(u, invokers));
37.             } else { // not a registry url
38.                 invoker = cluster.join(new StaticDirectory(invokers));
39.             }
40.         }
41.     }
43.     // 生成代理类
44.     return (T) proxyFactory.getProxy(invoker);
45. }

复制代码

createProxy 是服务引用时，生成服务代理对象的方法。这里会判断，如果有多个注册中心，会再封装一层集群，也就是先选择注册中心，再选择服务提供者。
这里一般情况 registryURL 不为空，cluster 使用的是AvailableCluster

1. public class AvailableCluster implements Cluster {
2.     public static final String NAME = "available";
4.     @Override
5.     public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
6.         return new AbstractClusterInvoker<T>(directory) {
7.             @Override
8.             public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
9.                 for (Invoker<T> invoker : invokers) {
10.                     if (invoker.isAvailable()) {
11.                         return invoker.invoke(invocation);
12.                     }
13.                 }
14.                 throw new RpcException("No provider available in " + invokers);
15.             }
16.         };
17.     }
18. }

复制代码

AvailableCluster的逻辑很简单，按顺序选择可使用的 invoker （这里的invoker其实就是每个注册中心）
总结

本篇文章介绍了Dubbo集群容错的整体工作过程和调用逻辑。Dubbo提供了多种集群实现，本文只介绍了Failover Cluster，其余实现感兴趣的可以自行查看源码。