Dubbo 的集群容错模式：Failback Cluster

本文主要是介绍Dubbo 的集群容错模式：Failback Cluster，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

集群容错系列文章：
Failover Cluster 失败自动切换
Failfast Cluster 快速失败，抛出异常
Failsafe Cluster 快速失败，不抛出异常
Failback Cluster 失败后定时重试
Forking Cluster 并行调用多个实例，只要一个成功就返回
Broadcast Cluster 广播调用所有实例，有一个报错则抛出异常
Available Cluster 可用的实例
Mergeable Cluster 合并结果

本文简单介绍 Dubbo 中的 Failback Cluster（失败自动回复，定时重发）。

简介

调用实例发生异常后，一段时间后重新再调用，直到调用成功。

如何使用

<dubbo:service cluster="failback" />

或

<dubbo:reference cluster="failback" />

实现逻辑

根据负载均衡算法选中被调用实例
执行选中的实例
执行成功则返回；执行有异常则定时重新发送请求，默认发送定时时间为 5 秒

实现代码

public class FailbackClusterInvoker<T> extends AbstractClusterInvoker<T> {private static final Logger logger = LoggerFactory.getLogger(FailbackClusterInvoker.class);private static final long RETRY_FAILED_PERIOD = 5 * 1000;/*** Use {@link NamedInternalThreadFactory} to produce {@link org.apache.dubbo.common.threadlocal.InternalThread}* which with the use of {@link org.apache.dubbo.common.threadlocal.InternalThreadLocal} in {@link RpcContext}.*/private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2,new NamedInternalThreadFactory("failback-cluster-timer", true));private final ConcurrentMap<Invocation, AbstractClusterInvoker<?>> failed = new ConcurrentHashMap<Invocation, AbstractClusterInvoker<?>>();private volatile ScheduledFuture<?> retryFuture;public FailbackClusterInvoker(Directory<T> directory) {super(directory);}private void addFailed(Invocation invocation, AbstractClusterInvoker<?> router) {if (retryFuture == null) {synchronized (this) {// 锁定当前对象if (retryFuture == null) {// 双重检查是否初始化实例 retryFutureretryFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {@Overridepublic void run() {// collect retry statisticstry {retryFailed();} catch (Throwable t) { // Defensive fault tolerancelogger.error("Unexpected error occur at collect statistic", t);}}// 5s 发送重试一次}, RETRY_FAILED_PERIOD, RETRY_FAILED_PERIOD, TimeUnit.MILLISECONDS);}}}failed.put(invocation, router);}// 重新调用void retryFailed() {// 没有调用失败的实例则不进行重试if (failed.size() == 0) {return;}// 遍历所有调用失败的实例，进行重试调用for (Map.Entry<Invocation, AbstractClusterInvoker<?>> entry : new HashMap<Invocation, AbstractClusterInvoker<?>>(failed).entrySet()) {Invocation invocation = entry.getKey();Invoker<?> invoker = entry.getValue();try {invoker.invoke(invocation);failed.remove(invocation);} catch (Throwable e) {logger.error("Failed retry to invoke method " + invocation.getMethodName() + ", waiting again.", e);}}}@Overrideprotected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {try {checkInvokers(invokers, invocation);// 根据负载均衡算法选中调用实例Invoker<T> invoker = select(loadbalance, invocation, invokers, null);// 执行调用实例return invoker.invoke(invocation);} catch (Throwable e) {logger.error("Failback to invoke method " + invocation.getMethodName() + ", wait for retry in background. Ignored exception: "+ e.getMessage() + ", ", e);// 调用出现异常，则定时重试直到成功addFailed(invocation, this);return new RpcResult(); // ignore}}}