Dubbo源码学习--FailbackCluster集群容错（四）

小鱼儿 2022-05-28 01:23 151阅读 0赞

### Failback Cluster ###

失败自动恢复，后台记录失败请求，定时重发。通常用于消息通知操作。

FailbackCluster的实现类如下：

public class FailbackCluster implements Cluster {
    
        public final static String NAME = "failback";
    
        public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
            return new FailbackClusterInvoker<T>(directory);
        }
    
    }

在FailbackClusterInvoker中完成请求失败定时重发机制，在FailbackClusterInvoker会起一个定时线程在线程池中不断进行重试操作。

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;
    
        private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2, new NamedThreadFactory("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) {
                        retryFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
    
                            public void run() {
                                // collect retry statistics
                                try {
                                    retryFailed();
                                } catch (Throwable t) { // Defensive fault tolerance
                                    logger.error("Unexpected error occur at collect statistic", t);
                                }
                            }
                        }, 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);
                }
            }
        }
    
        protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
            try {
                checkInvokers(invokers, invocation);
    			//通过负载均衡机制选择Invoker
                Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
    			//Invoker调用服务
                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
            }
        }
    
    }