接上篇《Dubbo技术知识总结之三——Dubbo 启动与服务暴露、引用》

四. 集群容错

在客户端已经从注册中心拉取和订阅服务列表完毕的前提下，Dubbo 完成一次完整的 RPC 调用，流程如下：

1. 服务列表聚合；
2. 路由；
3. 负载均衡；
4. 选择一台机器进行 RPC 调用；
5. 请求交给底层 I/O 线程池处理；
6. 读写、序列化、反序列化；
7. 方法调用；

将上面的步骤进行细化，在一次 RPC 调用过程中，Cluster 层的流程如下：

1. 根据不同的容错机制，生成 Invoker 对象，调用 AbstractClusterInvoker 的 Invoker 方法；
2. 获得可调用的服务列表；
3. 使用 Router 接口处理服务列表，根据路由规则过滤一部分服务；
4. 负载均衡；
5. RPC 调用；

其中步骤 1, 2, 3 是模板方法，使用通用的校验、参数准备等准备工作。最终，不同的容错机制的子类实现不同的 doInvoke 方法，每个子类方法都有各自的路由、负载均衡实现策略。

本章节主要总结 RPC 在 Cluster 层的工作，涉及步骤 1, 2, 3, 4，其中容错机制见5.1，容错过程中获取 Invoker 列表需要用到 Directory，见5.2；Directory 过程中需要用到路由，见5.3；负载均衡见5.4。剩余步骤 5, 6, 7 是具体的 RPC 调用，见第六章。

4.1 容错机制

容错过程是在各容错机制实现子类的 doInvoke 方法重写实现的。容错过程对上层用户是完全透明的，上层用户不用关心容错过程是怎么实现的，同时用户也可以通过不同的配置项来选择不同的容错机制。支持的容错机制如下：

注： 大部分容错机制的核心步骤都是：

1. 校验；
2. 获取配置参数；
3. 实现各自容错机制的调用；

在上述步骤 3 容错机制的调用中，主要步骤都是：

1. 校验；
2. 负载均衡；
3. RPC 调用；

如果有不同，在各自条目中进行说明

1. Failover：重试失败，默认策略
   • 调用失败，尝试调用其他服务器；
   • 根据配置的重试次数，进行重试；如果有成功，则返回；全部重试失败之后，抛出异常；
2. Failfast：快速失败
   • RPC 调用失败后，将异常封装为 RpcException，抛出并返回，不做任何重试；
3. Failsafe：安全失败
   • 出现异常时忽略；
4. Failback：定时重试失败
   • 调用失败后，将该失败的 invocation 缓存到 ConcurrentHashMap 中，并返回空结果集；同时设置定时线程池，定时时间到了就将失败的任务投入线程池，重新请求；
   • 如果重新请求成功，则从缓存中移除，请求失败则判断失败次数；如果失败次数少于设定的阈值，则重新投入定时线程池；如果多于设定的阈值，打印错误并放弃该请求；
   • 定时重试失败的实现思路，可以用于 Kafka 的重试队列；
5. Forking：并行
   • 根据设定的并行数量，循环执行负载均衡，筛选出可调用的 Invoker 列表；
   • 循环使用线程池，同时调用多个相同的服务；多个服务中，只要其中一个返回，就立即返回结果；所有线程调用失败，则抛出异常；
     • 该部分的实现是通过阻塞队列 BlockingQueue 实现的；将多个调用任务投入线程池后，任务执行结果投入 BlockingQueue；
     • 如果任务执行结果是异常类型，投入 BlockingQueue 抛出异常；此时记录异常次数，只有到记录异常次数等于服务数量时，说明所有服务都抛出异常，此时再将异常信息投入 BlockingQueue
     • 调用任务投入线程池之后，就立即调用 BlockingQueue # poll(int) 方法拉取结果，拉取到第一个结果就返回。如果返回值正常，就是其中一个服务的返回结果；如果返回值为 Exception 类型，说明所有服务都出现异常；
6. Broadcast：广播
   • 广播调用所有可用服务，循环遍历所有 Invoker，每个 Invoker 分别做 RPC 调用；
   • 如果有任意一个节点报错，等待广播最后完成之后抛出；如果多个节点异常，最后一个节点抛出的异常会覆盖前面抛出的异常；
7. Available：可用
   • 最简单的方式，请求不会做负载均衡，遍历所有服务列表，找到第一个可用节点，直接请求并返回结果；
8. Mock：仿真
   • 调用失败时返回伪造的响应结果，或者直接强行返回伪造结果；
9. Mergeable：合并：将多个节点请求的结果合并；

4.2 Directory

容错过程中需要获取 Invoker 列表，用于后续的路由和负载均衡。这个过程需要用到 Directory # list 方法执行。Directory 接口有一个抽象类 AbstractDirectory，以及两个主要实现类：动态列表 RegistryDirectory，以及静态列表 StaticDirectory。主要总结的是动态列表 RegistryDirectory，以及封装了基础方法的抽象类 AbstractDirectory。

RegistryDirectory 主要实现了两个功能：

1. 与注册中心的订阅，动态更新本地的 Invoker 列表；
2. 实现父类的 doList 方法；

4.2.1 订阅与动态更新

注册中心订阅的部分主要在 ZookeeperRegistry # doSubscribe() 方法中实现，见第二章注册中心部分。

在监听到注册中心对应 URL 变化后，触发 RegistryDirectory 对各种本地配置的动态更新。更新的配置包括：

1. 路由信息：通过路由工厂 RouterFactory 将 URL 包装成路由规则（见5.3），更新本地路由信息；
   • 更新路由规则，是通过 override 协议实现的；
2. 服务提供者配置 Configurator：管理员可以在 dubbo-admin 下动态修改生产者的参数，这些参数会保存在配置中心的 configurators 类目录下；
3. Invoker 修改：如果监听到的 Invoker 类型 URL 不为空，则将新的 URL 与本地旧 URL 合并，同时销毁旧 Invoker；

4.2.2 doList

doList 方法主要作用，就是调用路由方法。

4.3 路由

注：路由的整体思路与笔者设计的动态汇总统计业务不谋而合，通过表达式的方式实现数据的处理。

路由会根据用户配置的不同路由策略，对 Invoker 列表进行过滤。主要分为条件路由、文本路由、脚本路由。路由工厂 RouterFactory 是一个 SPI 接口，用户可以自行通过实现 Router 接口扩展 Router 类；在调用的时候，在 URL 的 protocol 参数中可以设置 file / script / condition，分别寻找对应的实现类。

4.3.1 条件路由 (ConditionRouter)

条件路由使用的是 condition://协议，URL 形式是：“condition://0.0.0.0/com.foo.DemoService?category=routers&dynamic=false&rule=” + URL.encode(“host = 10.20.153.10 => host = 10.20.153.11”)；每个参数都是有含义的：

条件路由最关键的部分在于 rule 的路由规则。以下面的路由规则为例：

method = find* => host = 192.168.1.22

1. 该路由规则的意义：所有调用 find 开头的方法，都会被路由到 192.168.1.22 的服务节点上；
2. => 之前部分是服务消费者匹配条件；
   • 如果匹配条件为空，则表示应用于所有消费者；
3. => 之后部分是服务提供者列表的过滤条件；
   • 如果过滤条件为空，则表示禁止访问；
4. 表示规则的表达式支持 $protocol 等占位符方式，也支持 =, != 等条件，也支持通配符 *。

条件路由的具体实现类是 ConditionRouter，整体的思想是通过正则表达式，按照 =>进行分割，然后对符号前后的内容进行正则表达式的匹配，匹配结果存入对象 MatchPair 中。对于上述的占位符、通配符等，MatchPair 会进行匹配解析。

注：条件路由的整体思路，类似于笔者设计的动态汇总统计业务。

4.3.2 文件路由 (FileRouter)

文件路由通常和脚本路由搭配使用。文件路由将规则写到文件中，文件中写的是自定义的脚本规则，脚本可以是 Javascript, Groovy 等，文件路由 FileRouter 找到对应文件，将文件中的脚本内容按照类型匹配脚本路由，执行解析。

4.3.3 脚本路由 (ScriptRouter)

脚本路由使用 JDK 自带的脚本解析器，对脚本解析并运行，默认使用 Javascript 解析器。在构造脚本路由时初始化脚本执行引擎，根据脚本不同的类型，通过 JDK 提供的 ScriptEngineManager 创建不同的脚本执行器。接收到脚本内容后，执行 route 方法。具体的过滤逻辑需要用户自行定义。

注：在笔者设计的动态汇总统计业务中，笔者使用了 Aviator 表达式引擎，它与脚本路由中的脚本执行器 ScriptEngineManager 类似。

4.4 负载均衡

很多容错策略在路由选择出所有可用 Invoker 列表中实行最后一步筛选，负载均衡。

负载均衡的核心是 LoadBalance 接口及其子类具体实现的，但并不是直接使用 LoadBalance 方法。在容错策略中的负载均衡先使用了抽象父类 AbstractClusterInvoker 中定义的 Invoker select 方法，它在 LoadBalance 基础上又封装了一些特性：

1. 粘滞连接：尽可能让客户端总是向同一提供者发起调用。
   • 类似的策略，也在 Kafka 再均衡策略 StickyAssignor 中用过；
2. 可用检测；
3. 避免重复调用；

select 方法也使用了模板模式，在 select 方法中处理通用逻辑，最后提供 doSelect 抽象方法供各子类具体实现。Dubbo 内置了四种负载均衡算法，此外由于 LoadBalance 接口带有 @SPI 注解，所以用户也可以自行扩展负载均衡算法。在调用方法时我们可以在 URL 中通过 loadbalance=xxx 动态指定 select 方法的负载均衡算法。

4.4.1 Random

根据权重，设置随机概率做负载均衡。

4.4.2 RoundRobin

见《Nginx》篇 6.2.2。

4.4.3 LeastActive

LeastActive 就是最少活跃调用负载均衡，Dubbo 在运行过程中会统计每一次 Invoker 的调用，每次从活跃数最少的 Invoker 中选一个节点。

4.4.4 一致性 Hash

一致性 Hash 的原理见《数据结构与算法》篇第五章。

Dubbo 的一致性 Hash 负载均衡，将接口名 + 方法名作为 Key 值，类型为 ConsistentHashSelector 实例对象作为 Value 存入一个 ConcurrentHashMap 中。每次请求进入，解析请求获取到方法，将该方法转为 Key 值，找到对应的 ConsistentHashSelector 进行负载均衡。所以 ConsistentHashSelector 是 Dubbo 中一致性 Hash 实现的核心。

ConsistentHashSelector 的环形散列是用 TreeMap 实现的，所有真实节点、虚拟节点都放在 TreeMap 中。将节点的 IP + 递增数字，然后作 MD5 计算，最后进行 Hash 计算，作为 TreeMap 的 Key 值。TreeMap 的 Value 值为对应的某个可以调用的节点。关键代码如下：

    // 遍历所有节点
    for (Invoker<T> invoker : invokers) {
        // 得到每个节点的 IP
        String address = invoker.getUrl().getAddress();
        // replicaNumber 是生成的虚拟节点数量，默认 160 个
        for (int i = 0; i < replicaNumber / 4; i++) {
            // 对 IP + 递增数字作 MD5 计算，作为节点标识
            byte[] digest = md5(address + i);
            for (int h = 0; h < 4; h++) {
                // 对标识作 Hash 计算，作为 TreeMap 的 Key 值
                long m = hash(digest, h);
                // 当前 Invoker 为 Value
                virtualInvokers.put(m, invoker);
            }
        }
    }

每次请求进来后，进行上述的 Key 值运算，每次请求的参数都不同，但是由于 TreeMap 是有序的树形结构，所以可以调用 TreeMap#ceilingEntry 方法，找到最近一个大于或等于给定 Key 值的节点 Entry。这样的操作相当于一致性 Hash 算法的顺时针向前查找的效果。