1.简介 为了避免单点故障，现在的应用至少会部署在两台服务器上。对于一些负载比较高的服务，会部署更多台服务器。这样，同一环境下的服务提供者数量会大于1。对于服务消费者来说，同一环境下出现了多个服务提供者。这时会出现一个问题，服务消费者需要决定选择哪个服务提供者进行调用。另外服务调用失败时的处理措施也是需要考虑的，是重试呢，还是抛出异常，亦或是只打印异常等。为了处理这些问题，Dubbo 定义了集群接口 Cluster 以及及 Cluster Invoker。集群 Cluster 用途是将多个服务提供者合并为一个 Cluster Invoker，并将这个 Invoker 暴露给服务消费者。这样一来，服务消费者只需通过这个 Invoker 进行远程调用即可，至于具体调用哪个服务提供者，以及调用失败后如何处理等问题，现在都交给集群模块去处理。集群模块是服务提供者和服务消费者的中间层，为服务消费者屏蔽了服务提供者的情况，这样服务消费者就可以处理远程调用相关事宜。比如发请求，接受服务提供者返回的数据等。这就是集群的作用。 Dubbo 提供了多种集群实现，包含但不限于 Failover Cluster、Failfast Cluster 和 Failsafe Cluster 等。每种集群实现类的用途不同，接下来我会一一进行分析。 2. 集群容错 在对集群相关代码进行分析之前，这里有必要先来介绍一下集群容错的所有组件。包含 Cluster、Cluster Invoker、Directory、Router 和 LoadBalance 等，先来看图。 * 图片来源：Dubbo 官方文档 这张图来自 Dubbo 官方文档，接下来我会按照这张图介绍集群工作过程。集群工作过程可分为两个阶段，第一个阶段是在服务消费者初始化期间，集群 Cluster 实现类为服务消费者创建 Cluster Invoker 实例，即上图中的 merge 操作。第二个阶段是在服务消费者进行远程调用时。以 FailoverClusterInvoker 为例，该类型 Cluster Invoker 首先会调用 Directory 的 list 方法列举 Invoker 列表（可将 Invoker 简单理解为服务提供者）。Directory 的用途是保存 Invoker，可简单类比为 List。其实现类 RegistryDirectory 是一个动态服务目录，可感知注册中心配置的变化，它所持有的 Inovker 列表会随着注册中心内容的变化而变化。每次变化后，RegistryDirectory 会动态增删 Inovker，并调用 Router 的 route 方法进行路由，过滤掉不符合路由规则的 Invoker。回到上图，Cluster Invoker 实际上并不会直接调用 Router 进行路由。当 FailoverClusterInvoker 拿到 Directory 返回的 Invoker 列表后，它会通过 LoadBalance 从 Invoker 列表中选择一个 Inovker。最后 FailoverClusterInvoker 会将参数传给 LoadBalance 选择出的 Invoker 实例的 invoker 方法，进行真正的 RPC 调用。 以上就是集群工作的整个流程，这里并没介绍集群是如何容错的。Dubbo 主要提供了这样几种容错方式： Failover Cluster - 失败自动切换 Failfast Cluster - 快速失败 Failsafe Cluster - 失败安全 Failback Cluster - 失败自动恢复 Forking Cluster - 并行调用多个服务提供者 这里暂时只对这几种容错模式进行简单的介绍，在接下来的章节中，我会重点分析这几种容错模式的具体实现。好了，关于集群的工作流程和容错模式先说到这，接下来进入源码分析阶段。 3.源码分析 3.1 Cluster 实现类分析 我在上一章提到了集群接口 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 Invoker join(Directory directory) throws RpcException { // 创建并返回 FailoverClusterInvoker 对象 return new FailoverClusterInvoker(directory); } } 如上，FailoverCluster 总共就包含这几行代码，用于创建 FailoverClusterInvoker 对象，很简单。下面再看一个。 public class FailbackCluster implements Cluster { public final static String NAME = "failback"; @Override public Invoker join(Directory directory) throws RpcException { // 创建并返回 FailbackClusterInvoker 对象 return new FailbackClusterInvoker(directory); } } 如上，FailbackCluster 的逻辑也是很简单，无需解释了。所以接下来，我们把重点放在各种 Cluster Invoker 上 3.2 Cluster Invoker 分析 我们首先从各种 Cluster Invoker 的父类 AbstractClusterInvoker 源码开始说起。前面说过，集群工作过程可分为两个阶段，第一个阶段是在服务消费者初始化期间，这个在服务引用那篇文章中已经分析过了，这里不再赘述。第二个阶段是在服务消费者进行远程调用时，此时 AbstractClusterInvoker 的 invoke 方法会被调用。列举 Invoker，负载均衡等操作均会在此阶段被执行。因此下面先来看一下 invoke 方法的逻辑。 public Result invoke(final Invocation invocation) throws RpcException { checkWhetherDestroyed(); LoadBalance loadbalance = null; // 绑定 attachments 到 invocation 中. Map contextAttachments = RpcContext.getContext().getAttachments(); if (contextAttachments != null && contextAttachments.size() != 0) { ((RpcInvocation) invocation).addAttachments(contextAttachments); } // 列举 Invoker List> 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> invokers, LoadBalance loadbalance) throws RpcException; AbstractClusterInvoker 的 invoke 方法主要用于列举 Invoker，以及加载 LoadBalance。最后再调用模板方法 doInvoke 进行后续操作。下面我们来看一下 Invoker 列举方法 list(Invocation) 的逻辑，如下： protected List> list(Invocation invocation) throws RpcException { // 调用 Directory 的 list 方法 List> invokers = directory.list(invocation); return invokers; } 如上，AbstractClusterInvoker 中的 list 方法做的事情很简单，只是简单的调用了 Directory 的 list 方法，没有其他更多的逻辑了。Directory 的 list 方法我在前面的文章中已经分析过了，这里就不赘述了。 接下来，我们把目光转移到 AbstractClusterInvoker 的各种实现类上，来看一下这些实现类是如何实现 doInvoke 方法逻辑的。 3.2.1 FailoverClusterInvoker FailoverClusterInvoker 在调用失败时，会自动切换 Invoker 进行重试。在无明确配置下，Dubbo 会使用这个类作为缺省 Cluster Invoker。下面来看一下该类的逻辑。 public class FailoverClusterInvoker extends AbstractClusterInvoker { // 省略部分代码 @Override public Result doInvoke(Invocation invocation, final List> invokers, LoadBalance loadbalance) throws RpcException { List> 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> invoked = new ArrayList>(copyinvokers.size()); Set providers = new HashSet(len); // 循环调用，失败重试 for (int i = 0; i < len; i++) { if (i > 0) { checkWhetherDestroyed(); // 在进行重试前重新列举 Invoker，这样做的好处是，如果某个服务挂了， // 通过调用 list 可得到最新可用的 Invoker 列表 copyinvokers = list(invocation); // 对 copyinvokers 进行判空检查 checkInvokers(copyinvokers, invocation); } // 通过负载均衡选择 Invoker Invoker 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 select(LoadBalance loadbalance, Invocation invocation, List> invokers, List> 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 对应的服务提供者可能因网络原因未能成功提供服务。 // 但是该提供者并没挂，此时 invokers 列表中仍存在该服务提供者对应的 Invoker。 if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) { // availablecheck 表示是否开启了可用性检查，如果开启了，则调用 stickyInvoker 的 // isAvailable 方法进行检查，如果检查通过，则直接返回 stickyInvoker。 if (availablecheck && stickyInvoker.isAvailable()) { return stickyInvoker; } } } // 如果线程走到当前代码处，说明前面的 stickyInvoker 为空，或者不可用。 // 此时调用继续调用 doSelect 选择 Invoker Invoker invoker = doSelect(loadbalance, invocation, invokers, selected); // 如果 sticky 为 true，则将负载均衡组件选出的 Invoker 赋值给 stickyInvoker if (sticky) { stickyInvoker = invoker; } return invoker; } 如上，select 方法的主要逻辑集中在了对粘滞连接特性的支持上。首先是获取 sticky 配置，然后再检测 invokers 列表中是否包含 stickyInvoker，如果不包含，则认为该 stickyInvoker 不可用，此时将其置空。这里的 invokers 列表可以看做是存活着的服务提供者列表，如果这个列表不包含 stickyInvoker，那自然而然的认为 stickyInvoker 挂了，所以置空。如果 stickyInvoker 存在于 invokers 列表中，此时要进行下一项检测 ---- 检测 selected 中是否包含 stickyInvoker。如果包含的话，说明 stickyInvoker 在此之前没有成功提供服务（但其仍然处于存活状态）。此时我们认为这个服务不可靠，不应该在重试期间内再次被调用，因此这个时候不会返回该 stickyInvoker。如果 selected 不包含 stickyInvoker，此时还需要进行可用性检测，比如检测服务提供者网络连通性等。当可用性检测通过，才可返回 stickyInvoker，否则调用 doSelect 方法选择 Invoker。如果 sticky 为 true，此时会将 doSelect 方法选出的 Invoker 赋值给 stickyInvoker。 以上就是 select 方法的逻辑，这段逻辑看起来不是很复杂，但是信息量比较大。不搞懂 invokers 和 selected 两个入参的含义，以及粘滞连接特性，这段代码应该是没法看懂的。大家在阅读这段代码时，不要忽略了对背景知识的理解。其他的不多说了，继续向下分析。 private Invoker doSelect(LoadBalance loadbalance, Invocation invocation, List> invokers, List> selected) throws RpcException { if (invokers == null || invokers.isEmpty()) return null; if (invokers.size() == 1) return invokers.get(0); if (loadbalance == null) { // 如果 loadbalance 为空，这里通过 SPI 加载 Loadbalance，默认为 RandomLoadBalance loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE); } // 通过负载均衡组件选择 Invoker Invoker invoker = loadbalance.select(invokers, getUrl(), invocation); // 如果 selected 包含负载均衡选择出的 Invoker，或者该 Invoker 无法经过可用性检查，此时进行重选 if ((selected != null && selected.contains(invoker)) || (!invoker.isAvailable() && getUrl() != null && availablecheck)) { try { // 进行重选 Invoker rinvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck); if (rinvoker != null) { // 如果 rinvoker 不为空，则将其赋值给 invoker invoker = rinvoker; } else { // rinvoker 为空，定位 invoker 在 invokers 中的位置 int index = invokers.indexOf(invoker); try { // 获取 index + 1 位置处的 Invoker，以下代码等价于： // invoker = invokers.get((index + 1) % invokers.size()); invoker = index < invokers.size() - 1 ? invokers.get(index + 1) : invokers.get(0); } catch (Exception e) { logger.warn("... may because invokers list dynamic change, ignore."); } } } catch (Throwable t) { logger.error("cluster reselect fail reason is : ..."); } } return invoker; } doSelect 主要做了两件事，第一是通过负载均衡组件选择 Invoker。第二是，如果选出来的 Invoker 不稳定，或不可用，此时需要调用 reselect 方法进行重选。若 reselect 选出来的 Invoker 为空，此时定位 invoker 在 invokers 列表中的位置 index，然后获取 index + 1 处的 invoker，这也可以看做是重选逻辑的一部分。关于负载均衡的选择逻辑，我将会在下篇文章进行详细分析。下面我们来看一下 reselect 方法的逻辑。 private Invoker reselect(LoadBalance loadbalance, Invocation invocation, List> invokers, List> selected, boolean availablecheck) throws RpcException { List> reselectInvokers = new ArrayList>(invokers.size() > 1 ? (invokers.size() - 1) : invokers.size()); // 根据 availablecheck 进行不同的处理 if (availablecheck) { // 遍历 invokers 列表 for (Invoker invoker : invokers) { // 检测可用性 if (invoker.isAvailable()) { // 如果 selected 列表不包含当前 invoker，则将其添加到 reselectInvokers 中 if (selected == null || !selected.contains(invoker)) { reselectInvokers.add(invoker); } } } // reselectInvokers 不为空，此时通过负载均衡组件进行选择 if (!reselectInvokers.isEmpty()) { return loadbalance.select(reselectInvokers, getUrl(), invocation); } // 不检查 Invoker 可用性 } else { for (Invoker invoker : invokers) { // 如果 selected 列表不包含当前 invoker，则将其添加到 reselectInvokers 中 if (selected == null || !selected.contains(invoker)) { reselectInvokers.add(invoker); } } if (!reselectInvokers.isEmpty()) { // 通过负载均衡组件进行选择 return loadbalance.select(reselectInvokers, getUrl(), invocation); } } { // 若线程走到此处，说明 reselectInvokers 集合为空，此时不会调用负载均衡组件进行筛选。 // 这里从 selected 列表中查找可用的 Invoker，并将其添加到 reselectInvokers 集合中 if (selected != null) { for (Invoker invoker : selected) { if ((invoker.isAvailable()) && !reselectInvokers.contains(invoker)) { reselectInvokers.add(invoker); } } } if (!reselectInvokers.isEmpty()) { // 再次进行选择，并返回选择结果 return loadbalance.select(reselectInvokers, getUrl(), invocation); } } return null; } reselect 方法总结下来其实只做了两件事情，第一是查找可用的 Invoker，并将其添加到 reselectInvokers 集合中。第二，如果 reselectInvokers 不为空，则通过负载均衡组件再次进行选择。其中第一件事情又可进行细分，一开始，reselect 从 invokers 列表中查找有效可用的 Invoker，若未能找到，此时再到 selected 列表中继续查找。关于 reselect 方法就先分析到这，继续分析