Dubbo 中的集群容错

前言

在微服务架构中，服务间的依赖关系复杂且动态，任何一个服务的故障都可能引发连锁反应，导致系统雪崩。一个好的容错设计可以避免这些问题发生：

• 服务雪崩效应：单个服务崩溃或响应延迟可能导致调用链上的所有服务被阻塞，最终拖垮整个系统。例如，若服务 A 依赖服务 B，而服务 B 因高负载无法响应，A 的线程池可能被占满，进而影响其他依赖A的服务；
  
• 分布式系统的脆弱性：网络抖动、节点宕机、资源耗尽等问题在分布式环境中不可避免。容错机制通过冗余和快速失败策略，确保部分故障不会扩散到整个系统；
  
• 服务的可用性低：微服务的目标是提升系统可用性，而容错设计（如故障转移、熔断）是保障服务持续可用的核心手段。例如，通过自动切换健康节点，避免单点故障。
  

Dubbo 的集群容错机制

在 Dubbo 中，多个 Provider 实例构成一个「集群」。消费者调用时，Dubbo 通过 Cluster 模块实现容错策略的封装和路由，Cluster 模块会根据配置（如 cluster=failover）装配不同的容错策略实现类，对 Directory 中的多个 Invoker 进行处理，返回一个可执行的 Invoker。Dubbo 当前已支持以下 6 种容错策略（在 org.apache.dubbo.rpc.cluster.support 包下）：
策略简称实现类名特性使用场景FailoverFailoverClusterInvoker失败自动重试，默认实现网络不稳定，民登操作FailfastFailfastClusterInvoker快速失败，不重试响应时间敏感，非幂等FailsafeFailsafeClusterInvoker失败忽略异常日志记录、监控等非主要场景FailbackFailbackClusterInvoker失败后后台重试可容忍失败，后续补偿重试ForkingForkingClusterInvoker并行调用多个节点，最快成功返回实时性要求高，资源充足BroadcastBroadcastClusterInvoker广播方式调用所有服务提供着配置更新、通知类等操作Failover Cluster（失败自动切换，默认策略）

实现原理：通过循环重试实现容错。
实现源码关键点：

• FailoverClusterInvoker 的 doInvoke 方法中，通过 for 循环控制重试次数（默认重试 2 次，共调用 3 次）;
  
• 每次重试前调用 list(invocation) 重新获取最新的 Invoker 列表，确保动态感知节点变化。
  

1. // 代码片段：org.apache.dubbo.rpc.cluster.support.FailoverClusterInvoker#doInvoke
2. for (int i = 0; i < len; i++) {
3.     if (i > 0) {
4.         copyInvokers = list(invocation); // 动态刷新 Invoker 列表
5.     }
6.     Invoker<T> invoker = select(loadbalance, invocation, copyInvokers, invoked);
7.     // 调用并处理异常...
8. }

复制代码

Failfast Cluster（快速失败）

实现原理：仅发起一次调用，异常直接抛出。
实现源码关键点：

• FailfastClusterInvoker 直接调用目标 Invoker，不进行重试。
  

1. // 代码片段：org.apache.dubbo.rpc.cluster.support.FailfastClusterInvoker#doInvoke
2. fpublic Result doInvoke(...) throws RpcException {
3.     checkInvokers(invokers, invocation);
4.     Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
5.     return invoker.invoke(invocation); // 仅一次调用
6. }

复制代码

Failsafe Cluster（失败安全）

实现原理：异常被捕获后返回空结果，不中断流程。
实现源码关键点：

• ailsafeClusterInvoker通过try-catch捕获异常并记录日志。
  

1. // 代码片段：org.apache.dubbo.rpc.cluster.support.FailsafeClusterInvoker
2. try {
3.     // 调用逻辑...
4. } catch (Throwable e) {
5.     logger.error("Failsafe ignore exception", e);
6.     return new RpcResult(); // 返回空结果
7. }

复制代码

Failback Cluster（失败自动恢复）

实现原理：失败请求存入队列，定时重试。
实现源码关键点：

• 捕获失败异常，使用 RetryTimerTask 存储失败请求，定时触发重试。
  

1. // 代码片段：org.apache.dubbo.rpc.cluster.support.FailbackClusterInvoker#doInvoke
2. private void addFailed(
3.         LoadBalance loadbalance,
4.         Invocation invocation,
5.         List<Invoker<T>> invokers,
6.         Invoker<T> lastInvoker,
7.         URL consumerUrl) {
8.     if (failTimer == null) {
9.         synchronized (this) {
10.             if (failTimer == null) {
11.                 failTimer = new HashedWheelTimer(
12.                         new NamedThreadFactory("failback-cluster-timer", true),
13.                         1,
14.                         TimeUnit.SECONDS,
15.                         32,
16.                         failbackTasks);
17.             }
18.         }
19.     }
20.     RetryTimerTask retryTimerTask = new RetryTimerTask(
21.             loadbalance, invocation, invokers, lastInvoker, retries, RETRY_FAILED_PERIOD, consumerUrl);
22.     try {
23.         failTimer.newTimeout(retryTimerTask, RETRY_FAILED_PERIOD, TimeUnit.SECONDS);
24.     } catch (Throwable e) {
25.         logger.error(
26.                 CLUSTER_TIMER_RETRY_FAILED,
27.                 "add newTimeout exception",
28.                 "",
29.                 "Failback background works error, invocation->" + invocation + ", exception: " + e.getMessage(),
30.                 e);
31.     }
32. }

复制代码

Forking Cluster（并行调用）

实现原理：并发调用多个节点，首个成功结果即返回。
实现源码关键点：

• 使用线程池并发调用，结果通过 BlockingQueue 异步接收。
  

1. // 代码片段：org.apache.dubbo.rpc.cluster.support.ForkingClusterInvoker#doInvoke
2. for (Invoker<T> invoker : selected) {
3.     executor.execute(() -> {
4.         Result result = invoker.invoke(invocation);
5.         ref.offer(result); // 结果存入队列
6.     });
7. }

复制代码

Broadcast Cluster（广播调用）

实现原理：逐个调用所有节点，任一失败则整体失败。
实现源码关键点：

• 遍历所有 Invoker 调用，异常累积后抛出。
  

1. // 代码片段：org.apache.dubbo.rpc.cluster.support.BroadcastClusterInvoker#doInvoke
2. for (Invoker<T> invoker : invokers) {
3.     try {
4.         invoker.invoke(invocation);
5.     } catch (RpcException e) {
6.         exception = e;
7.     }
8. }
9. if (exception != null) throw exception;

复制代码

如何自定义集群容错策略

如果以上提供的容错策略不满足需求，Dubbo 支持通过 SPI 自定义 Cluster 实现，步骤如下：
第一步：实现 Cluster 和 AbstractClusterInvoker

1. @SPI("custom")
2. public class MyCluster implements Cluster {
4.     @Override
5.     public <T> Invoker<T> join(Directory<T> directory) {
6.         return new MyClusterInvoker<>(directory);
7.     }
9. }

复制代码

1. public class MyClusterInvoker<T> extends AbstractClusterInvoker<T> {
3.     @Override
4.     protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) {
5.         // 自定义逻辑，例如条件重试、动态路由等
6.     }
8. }

复制代码

第二步：添加 SPI 配置

在 META-INF/dubbo/org.apache.dubbo.rpc.cluster.Cluster 中添加配置：

1. mycluster=com.example.MyCluster

复制代码

第三步：配置使用自定义容错策略

[code][/code]总结

建议核心服务优先使用 Failover（失败自动切换） 策略保障可用性，非核心服务可降级为 Failsafe（失败安全）。同时结合 Hystrix(已停止更新) 或 Sentinel 实现熔断与限流，增强容错能力。
通过灵活组合 Dubbo 的容错策略，可显著提升分布式系统的鲁棒性。实际应用配置时需要根据业务特性权衡延迟、资源开销与一致性要求，一切皆是 trade off ～
P.S. 不妨再深入思考一下：Dubbo 的集群容错实现中有哪些优秀设计值得我们学习？

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！