37 集群容错：一个好汉三个帮（上） 你好，我是杨四正，今天我和你分享的主题是集群容错：一个好汉三个帮（上篇）。
在前面的课时中，我们已经对 Directory、Router、LoadBalance 等概念进行了深入的剖析，本课时将重点分析 Cluster 接口的相关内容。
Cluster 接口提供了我们常说的集群容错功能。
集群中的单个节点有一定概率出现一些问题，例如，磁盘损坏、系统崩溃等，导致节点无法对外提供服务，因此在分布式 RPC 框架中，必须要重视这种情况。为了避免单点故障，我们的 Provider 通常至少会部署在两台服务器上，以集群的形式对外提供服务，对于一些负载比较高的服务，则需要部署更多 Provider 来抗住流量。
在 Dubbo 中，通过 Cluster 这个接口把一组可供调用的 Provider 信息组合成为一个统一的 Invoker 供调用方进行调用。经过 Router 过滤、LoadBalance 选址之后，选中一个具体 Provider 进行调用，如果调用失败，则会按照集群的容错策略进行容错处理。
Dubbo 默认内置了若干容错策略，并且每种容错策略都有自己独特的应用场景，我们可以通过配置选择不同的容错策略。如果这些内置容错策略不能满足需求，我们还可以通过自定义容错策略进行配置。
了解了上述背景知识之后，下面我们就正式开始介绍 Cluster 接口。
Cluster 接口与容错机制
Cluster 的工作流程大致可以分为两步（如下图所示）：①创建 Cluster Invoker 实例（在 Consumer 初始化时，Cluster 实现类会创建一个 Cluster Invoker 实例，即下图中的 merge 操作）；②使用 Cluster Invoker 实例（在 Consumer 服务消费者发起远程调用请求的时候，Cluster Invoker 会依赖前面课时介绍的 Directory、Router、LoadBalance 等组件得到最终要调用的 Invoker 对象）。
Cluster 核心流程图
Cluster Invoker 获取 Invoker 的流程大致可描述为如下：
- 通过 Directory 获取 Invoker 列表，以 RegistryDirectory 为例，会感知注册中心的动态变化，实时获取当前 Provider 对应的 Invoker 集合；
- 调用 Router 的 route() 方法进行路由，过滤掉不符合路由规则的 Invoker 对象；
- 通过 LoadBalance 从 Invoker 列表中选择一个 Invoker；
- ClusterInvoker 会将参数传给 LoadBalance 选择出的 Invoker 实例的 invoke 方法，进行真正的远程调用。
这个过程是一个正常流程，没有涉及容错处理。Dubbo 中常见的容错方式有如下几个。
- Failover Cluster：失败自动切换。它是 Dubbo 的默认容错机制，在请求一个 Provider 节点失败的时候，自动切换其他 Provider 节点，默认执行 3 次，适合幂等操作。当然，重试次数越多，在故障容错的时候带给 Provider 的压力就越大，在极端情况下甚至可能造成雪崩式的问题。
- Failback Cluster：失败自动恢复。失败后记录到队列中，通过定时器重试。
- Failfast Cluster：快速失败。请求失败后返回异常，不进行任何重试。
- Failsafe Cluster：失败安全。请求失败后忽略异常，不进行任何重试。
- Forking Cluster：并行调用多个 Provider 节点，只要有一个成功就返回。
- Broadcast Cluster：广播多个 Provider 节点，只要有一个节点失败就失败。
- Available Cluster：遍历所有的 Provider 节点，找到每一个可用的节点，就直接调用。如果没有可用的 Provider 节点，则直接抛出异常。
- Mergeable Cluster：请求多个 Provider 节点并将得到的结果进行合并。
下面我们再来看 Cluster 接口。Cluster 接口是一个扩展接口，通过 @SPI 注解的参数我们知道其使用的默认实现是 FailoverCluster，它只定义了一个 join() 方法，在其上添加了 @Adaptive 注解，会动态生成适配器类，其中会优先根据 Directory.getUrl() 方法返回的 URL 中的 cluster 参数值选择扩展实现，若无 cluster 参数则使用默认的 FailoverCluster 实现。Cluster 接口的具体定义如下所示：
@SPI(FailoverCluster.NAME) public interface Cluster { @Adaptive
Cluster 接口的实现类如下图所示，分别对应前面提到的多种容错策略：
Cluster 接口继承关系
在每个 Cluster 接口实现中，都会创建对应的 Invoker 对象，这些都继承自 AbstractClusterInvoker 抽象类，如下图所示：
AbstractClusterInvoker 继承关系图
通过上面两张继承关系图我们可以看出，Cluster 接口和 Invoker 接口都会有相应的抽象实现类，这些抽象实现类都实现了一些公共能力。下面我们就来深入介绍 AbstractClusterInvoker 和 AbstractCluster 这两个抽象类。
AbstractClusterInvoker
了解了 Cluster Invoker 的继承关系之后，我们首先来看 AbstractClusterInvoker，它有两点核心功能：一个是实现的 Invoker 接口，对 Invoker.invoke() 方法进行通用的抽象实现；另一个是实现通用的负载均衡算法。
在 AbstractClusterInvoker.invoke() 方法中，会通过 Directory 获取 Invoker 列表，然后通过 SPI 初始化 LoadBalance，最后调用 doInvoke() 方法执行子类的逻辑。在 Directory.list() 方法返回 Invoker 集合之前，已经使用 Router 进行了一次筛选，你可以回顾前面[第 31 课时]对 RegistryDirectory 的分析。
public Result invoke(final Invocation invocation) throws RpcException { // 检测当前Invoker是否已销毁 checkWhetherDestroyed(); // 将RpcContext中的attachment添加到Invocation中 Map<String, Object> contextAttachments = RpcContext.getContext().getObjectAttachments(); if (contextAttachments != null && contextAttachments.size() != 0) { ((RpcInvocation) invocation).addObjectAttachments(contextAttachments); } // 通过Directory获取Invoker对象列表，通过对RegistryDirectory的介绍我们知道，其中已经调用了Router进行过滤 List<Invoker
下面我们来看一下 AbstractClusterInvoker 是如何按照不同的 LoadBalance 算法从 Invoker 集合中选取最终 Invoker 对象的。
AbstractClusterInvoker 并没有简单粗暴地使用 LoadBalance.select() 方法完成负载均衡，而是做了进一步的封装，具体实现在 select() 方法中。在 select() 方法中会根据配置决定是否开启粘滞连接特性，如果开启了，则需要将上次使用的 Invoker 缓存起来，只要 Provider 节点可用就直接调用，不会再进行负载均衡。如果调用失败，才会重新进行负载均衡，并且排除已经重试过的 Provider 节点。
// 第一个参数是此次使用的LoadBalance实现，第二个参数Invocation是此次服务调用的上下文信息， // 第三个参数是待选择的Invoker集合，第四个参数用来记录负载均衡已经选出来、尝试过的Invoker集合 protected Invoker
doSelect() 方法主要做了两件事：
- 一是通过 LoadBalance 选择 Invoker 对象；
- 二是如果选出来的 Invoker 不稳定或不可用，会调用 reselect() 方法进行重选。
private Invoker
doSelect(LoadBalance loadbalance, Invocation invocation, List<Invoker > invokers, List<Invoker > selected) throws RpcException { // 判断是否需要进行负载均衡，Invoker集合为空，直接返回null if (CollectionUtils.isEmpty(invokers)) { return null; } if (invokers.size() == 1) { // 只有一个Invoker对象，直接返回即可 return invokers.get(0); } // 通过LoadBalance实现选择Invoker对象 Invoker invoker = loadbalance.select(invokers, getUrl(), invocation); // 如果LoadBalance选出的Invoker对象，已经尝试过请求了或不可用，则需要调用reselect()方法重选 if ((selected != null && selected.contains(invoker)) // Invoker已经尝试调用过了，但是失败了 || (!invoker.isAvailable() && getUrl() != null && availablecheck) // Invoker不可用 ) { try { // 调用reselect()方法重选 Invoker rInvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck); // 如果重选的Invoker对象不为空，则直接返回这个 rInvoker if (rInvoker != null) { invoker = rInvoker; } else { int index = invokers.indexOf(invoker); try { // 如果重选的Invoker对象为空，则返回该Invoker的下一个Invoker对象 invoker = invokers.get((index + 1) % invokers.size()); } catch (Exception e) { logger.warn("..."); } } } catch (Throwable t) { logger.error("..."); } } return invoker; }
reselect() 方法会重新进行一次负载均衡，首先对未尝试过的可用 Invokers 进行负载均衡，如果已经全部重试过了，则将尝试过的 Provider 节点过滤掉，然后在可用的 Provider 节点中重新进行负载均衡。
private Invoker
AbstractCluster
常用的 ClusterInvoker 实现都继承了 AbstractClusterInvoker 类型，对应的 Cluster 扩展实现都继承了 AbstractCluster 抽象类。AbstractCluster 抽象类的核心逻辑是在 ClusterInvoker 外层包装一层 ClusterInterceptor，从而实现类似切面的效果。
下面是 ClusterInterceptor 接口的定义： @SPI public interface ClusterInterceptor { // 前置拦截方法 void before(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation); // 后置拦截方法 void after(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation); // 调用ClusterInvoker的invoke()方法完成请求 default Result intercept(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation) throws RpcException { return clusterInvoker.invoke(invocation); } // 这个Listener用来监听请求的正常结果以及异常 interface Listener { void onMessage(Result appResponse, AbstractClusterInvoker<?> clusterInvoker, Invocation invocation); void onError(Throwable t, AbstractClusterInvoker<?> clusterInvoker, Invocation invocation); } }
在 AbstractCluster 抽象类的 join() 方法中，首先会调用 doJoin() 方法获取最终要调用的 Invoker 对象，doJoin() 是个抽象方法，由 AbstractCluster 子类根据具体的策略进行实现。之后，AbstractCluster.join() 方法会调用 buildClusterInterceptors() 方法加载 ClusterInterceptor 扩展实现类，对 Invoker 对象进行包装。具体实现如下：
private
InterceptorInvokerNode 会将底层的 AbstractClusterInvoker 对象以及关联的 ClusterInterceptor 对象封装到一起，还会维护一个 next 引用，指向下一个 InterceptorInvokerNode 对象。
在 InterceptorInvokerNode.invoke() 方法中，会先调用 ClusterInterceptor 的前置逻辑，然后执行 intercept() 方法调用 AbstractClusterInvoker 的 invoke() 方法完成远程调用，最后执行 ClusterInterceptor 的后置逻辑。具体实现如下： public Result invoke(Invocation invocation) throws RpcException { Result asyncResult; try { interceptor.before(next, invocation); // 前置逻辑 // 执行invoke()方法完成远程调用 asyncResult = interceptor.intercept(next, invocation); } catch (Exception e) { if (interceptor instanceof ClusterInterceptor.Listener) { // 出现异常时，会触发监听器的onError()方法 ClusterInterceptor.Listener listener = (ClusterInterceptor.Listener) interceptor; listener.onError(e, clusterInvoker, invocation); } throw e; } finally { // 执行后置逻辑 interceptor.after(next, invocation); } return asyncResult.whenCompleteWithContext((r, t) -> { if (interceptor instanceof ClusterInterceptor.Listener) { ClusterInterceptor.Listener listener = (ClusterInterceptor.Listener) interceptor; if (t == null) { // 正常返回时，会调用onMessage()方法触发监听器 listener.onMessage(r, clusterInvoker, invocation); } else { listener.onError(t, clusterInvoker, invocation); } } }); }
Dubbo 提供了两个 ClusterInterceptor 实现类，分别是 ConsumerContextClusterInterceptor 和 ZoneAwareClusterInterceptor，如下图所示：
ClusterInterceptor 继承关系图
在 ConsumerContextClusterInterceptor 的 before() 方法中，会在 RpcContext 中设置当前 Consumer 地址、此次调用的 Invoker 等信息，同时还会删除之前与当前线程绑定的 Server Context。在 after() 方法中，会删除本地 RpcContext 的信息。ConsumerContextClusterInterceptor 的具体实现如下： public void before(AbstractClusterInvoker<?> invoker, Invocation invocation) { // 获取当前线程绑定的RpcContext RpcContext context = RpcContext.getContext(); // 设置Invoker、Consumer地址等信息 context.setInvocation(invocation).setLocalAddress(NetUtils.getLocalHost(), 0); if (invocation instanceof RpcInvocation) { ((RpcInvocation) invocation).setInvoker(invoker); } RpcContext.removeServerContext(); } public void after(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation) { RpcContext.removeContext(true); // 删除本地RpcContext的信息 }
ConsumerContextClusterInterceptor 同时继承了 ClusterInterceptor.Listener 接口，在其 onMessage() 方法中，会获取响应中的 attachments 并设置到 RpcContext 中的 SERVER_LOCAL 之中，具体实现如下：
public void onMessage(Result appResponse, AbstractClusterInvoker<?> invoker, Invocation invocation) { // 从AppResponse中获取attachment，并设置到SERVER_LOCAL这个RpcContext中 RpcContext.getServerContext().setObjectAttachments(appResponse.getObjectAttachments()); }
介绍完 ConsumerContextClusterInterceptor，我们再来看 ZoneAwareClusterInterceptor。
在 ZoneAwareClusterInterceptor 的 before() 方法中，会从 RpcContext 中获取多注册中心相关的参数并设置到 Invocation 中（主要是 registry_zone 参数和 registry_zone_force 参数，这两个参数的具体含义，在后面分析 ZoneAwareClusterInvoker 时详细介绍），ZoneAwareClusterInterceptor 的 after() 方法为空实现。ZoneAwareClusterInterceptor 的具体实现如下：
public void before(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation) { RpcContext rpcContext = RpcContext.getContext(); // 从RpcContext中获取registry_zone参数和registry_zone_force参数 String zone = (String) rpcContext.getAttachment(REGISTRY_ZONE); String force = (String) rpcContext.getAttachment(REGISTRY_ZONE_FORCE); // 检测用户是否提供了ZoneDetector接口的扩展实现 ExtensionLoader
需要注意的是，ZoneAwareClusterInterceptor 没有实现 ClusterInterceptor.Listener 接口，也就是不提供监听响应的功能。
总结
本课时我们主要介绍的是 Dubbo Cluster 层中容错机制相关的内容。首先，我们了解了集群容错机制的作用。然后，我们介绍了 Cluster 接口的定义以及其各个实现类的核心功能。之后，我们深入讲解了 AbstractClusterInvoker 的实现，其核心是实现了一套通用的负载均衡算法。最后，我们还分析了 AbstractCluster 抽象实现类以及其中涉及的 ClusterInterceptor 接口的内容。
参考资料
https://learn.lianglianglee.com/%e4%b8%93%e6%a0%8f/Dubbo%e6%ba%90%e7%a0%81%e8%a7%a3%e8%af%bb%e4%b8%8e%e5%ae%9e%e6%88%98-%e5%ae%8c/37%20%20%e9%9b%86%e7%be%a4%e5%ae%b9%e9%94%99%ef%bc%9a%e4%b8%80%e4%b8%aa%e5%a5%bd%e6%b1%89%e4%b8%89%e4%b8%aa%e5%b8%ae%ef%bc%88%e4%b8%8a%ef%bc%89.md
更多学习
更多实时资讯，前沿技术，生活趣事。尽在【老马啸西风】
交流社群：[交流群信息](https://mp.weixin.qq.com/s/rkSvXxiiLGjl3S-ZOZCr0Q)
