前言

在Dubbo——集群容錯(上)，介紹了 Dubbo Cluster 層中集群容錯機制的基礎知識，還說明了 Cluster 接口的定義以及其各個實現(xiàn)類的核心功能爹土。同時知举，還分析了 AbstractClusterInvoker 抽象類以及 AbstractCluster 抽象實現(xiàn)類的核心實現(xiàn)至朗。

那本文孕讳，將介紹 Cluster 接口的全部實現(xiàn)類奇徒，以及相關的 Cluster Invoker 實現(xiàn)類尉间。

FailoverClusterInvoker

通過前面對 Cluster 接口的介紹知道偿乖，Cluster 默認的擴展實現(xiàn)是 FailoverCluster，其 doJoin() 方法中會創(chuàng)建一個 FailoverClusterInvoker 對象并返回哲嘲，具體實現(xiàn)如下：

public class FailoverCluster extends AbstractCluster {

    public final static String NAME = "failover";

    @Override
    public <T> AbstractClusterInvoker<T> doJoin(Directory<T> directory) throws RpcException {
        return new FailoverClusterInvoker<>(directory);
    }

}

FailoverClusterInvoker 會在調(diào)用失敗的時候贪薪，自動切換 Invoker 進行重試。下面來看 FailoverClusterInvoker 的核心實現(xiàn)：

public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> {

    @Override
    @SuppressWarnings({"unchecked", "rawtypes"})
    public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        List<Invoker<T>> copyInvokers = invokers;
        // 檢查copyInvokers集合是否為空眠副，如果為空會拋出異常
        checkInvokers(copyInvokers, invocation);
        String methodName = RpcUtils.getMethodName(invocation);
        // 參數(shù)重試次數(shù)画切，默認重試2次，總共執(zhí)行3次
        int len = getUrl().getMethodParameter(methodName, RETRIES_KEY, DEFAULT_RETRIES) + 1;
        if (len <= 0) {
            len = 1;
        }
        // retry loop.
        RpcException le = null; // last exception.
        // 記錄已經(jīng)嘗試調(diào)用過的Invoker對象
        List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyInvokers.size()); // invoked invokers.
        Set<String> providers = new HashSet<String>(len);
        for (int i = 0; i < len; i++) {
            // 第一次傳進來的invokers已經(jīng)check過了囱怕，第二次則是重試霍弹，需要重新獲取最新的服務列表
            if (i > 0) {
                checkWhetherDestroyed();
                // 這里會重新調(diào)用Directory.list()方法，獲取Invoker列表
                copyInvokers = list(invocation);
                // 檢查copyInvokers集合是否為空娃弓，如果為空會拋出異常
                checkInvokers(copyInvokers, invocation);
            }
            // 通過LoadBalance選擇Invoker對象典格，這里傳入的invoked集合，
            // 就是前面介紹AbstractClusterInvoker.select()方法中的selected集合
            Invoker<T> invoker = select(loadbalance, invocation, copyInvokers, invoked);
            // 記錄此次要嘗試調(diào)用的Invoker對象台丛，下一次重試時就會過濾這個服務
            invoked.add(invoker);
            RpcContext.getContext().setInvokers((List) invoked);
            try {
                // 調(diào)用目標Invoker對象的invoke()方法耍缴，完成遠程調(diào)用
                Result result = invoker.invoke(invocation);
                // 經(jīng)過嘗試之后，終于成功，這里會打印一個警告日志防嗡，將嘗試過來的Provider地址打印出來
                if (le != null && logger.isWarnEnabled()) {
                    logger.warn("Although retry the method " + methodName
                            + " in the service " + getInterface().getName()
                            + " was successful by the provider " + invoker.getUrl().getAddress()
                            + ", but there have been failed providers " + providers
                            + " (" + providers.size() + "/" + copyInvokers.size()
                            + ") from the registry " + directory.getUrl().getAddress()
                            + " on the consumer " + NetUtils.getLocalHost()
                            + " using the dubbo version " + Version.getVersion() + ". Last error is: "
                            + le.getMessage(), le);
                }
                return result;
            } catch (RpcException e) {
                if (e.isBiz()) { // biz exception.
                    throw e;
                }
                le = e;
            } catch (Throwable e) {// 拋出異常变汪，表示此次嘗試失敗，會進行重試
                le = new RpcException(e.getMessage(), e);
            } finally {
                // 記錄嘗試過的Provider地址本鸣，會在上面的警告日志中打印出來
                providers.add(invoker.getUrl().getAddress());
            }
        }
        // 達到重試次數(shù)上限之后疫衩，會拋出異常，其中會攜帶調(diào)用的方法名荣德、嘗試過的Provider節(jié)點的地址(providers集合)闷煤、
        // 全部的Provider個數(shù)(copyInvokers集合)以及Directory信息
        throw new RpcException(le.getCode(), "Failed to invoke the method "
                + methodName + " in the service " + getInterface().getName()
                + ". Tried " + len + " times of the providers " + providers
                + " (" + providers.size() + "/" + copyInvokers.size()
                + ") from the registry " + directory.getUrl().getAddress()
                + " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version "
                + Version.getVersion() + ". Last error is: "
                + le.getMessage(), le.getCause() != null ? le.getCause() : le);
    }

}

FailbackClusterInvoker

FailbackCluster 是 Cluster 接口的另一個擴展實現(xiàn)，擴展名是 failback涮瞻，其 doJoin() 方法中創(chuàng)建的 Invoker 對象是 FailbackClusterInvoker 類型鲤拿，具體實現(xiàn)如下：

public class FailbackCluster extends AbstractCluster {

    public final static String NAME = "failback";

    @Override
    public <T> AbstractClusterInvoker<T> doJoin(Directory<T> directory) throws RpcException {
        return new FailbackClusterInvoker<>(directory);
    }

}

FailbackClusterInvoker 在請求失敗之后，返回一個空結果給 Consumer署咽，同時還會添加一個定時任務對失敗的請求進行重試近顷。下面來看 FailbackClusterInvoker 的具體實現(xiàn)：

public class FailbackClusterInvoker<T> extends AbstractClusterInvoker<T> {

    @Override
    protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        Invoker<T> invoker = null;
        try {
            // 檢測Invoker集合是否為空
            checkInvokers(invokers, invocation);
            // 調(diào)用select()方法得到此次嘗試的Invoker對象
            invoker = select(loadbalance, invocation, invokers, null);
            // 調(diào)用invoke()方法完成遠程調(diào)用
            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(loadbalance, invocation, invokers, invoker);
            return AsyncRpcResult.newDefaultAsyncResult(null, null, invocation); // ignore
        }
    }
}

在 doInvoke() 方法中宁否，請求失敗時會調(diào)用 addFailed() 方法添加定時任務進行重試窒升，默認每隔 5 秒執(zhí)行一次，總共重試 3 次慕匠，具體實現(xiàn)如下：

public class FailbackClusterInvoker<T> extends AbstractClusterInvoker<T> {

    private void addFailed(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, Invoker<T> lastInvoker) {
        if (failTimer == null) {
            synchronized (this) {
                // Double Check防止并發(fā)問題 
                if (failTimer == null) {
                    // 初始化時間輪饱须，這個時間輪有32個槽，每個槽代表1秒
                    failTimer = new HashedWheelTimer(
                            new NamedThreadFactory("failback-cluster-timer", true),
                            1,
                            TimeUnit.SECONDS, 32, failbackTasks);
                }
            }
        }
        // 創(chuàng)建一個定時任務
        RetryTimerTask retryTimerTask = new RetryTimerTask(loadbalance, invocation, invokers, lastInvoker, retries, RETRY_FAILED_PERIOD);
        try {
            // 將定時任務添加到時間輪中
            failTimer.newTimeout(retryTimerTask, RETRY_FAILED_PERIOD, TimeUnit.SECONDS);
        } catch (Throwable e) {
            logger.error("Failback background works error,invocation->" + invocation + ", exception: " + e.getMessage());
        }
    }
}

在 RetryTimerTask 定時任務中台谊，會重新調(diào)用 select() 方法篩選合適的 Invoker 對象蓉媳，并嘗試進行請求。如果請求再次失敗且重試次數(shù)未達到上限锅铅，則調(diào)用 rePut() 方法再次添加定時任務酪呻，等待進行重試；如果請求成功盐须，也不會返回任何結果玩荠。RetryTimerTask 的核心實現(xiàn)如下：

public class FailbackClusterInvoker<T> extends AbstractClusterInvoker<T> {

    private class RetryTimerTask implements TimerTask {
        private final Invocation invocation;
        private final LoadBalance loadbalance;
        private final List<Invoker<T>> invokers;
        private final int retries;
        private final long tick;
        private Invoker<T> lastInvoker;
        private int retryTimes = 0;
        
        @Override
        public void run(Timeout timeout) {
            try {
                // 重新選擇Invoker對象，注意贼邓，這里會將上次重試失敗的Invoker作為selected集合傳入
                Invoker<T> retryInvoker = select(loadbalance, invocation, invokers, Collections.singletonList(lastInvoker));
                lastInvoker = retryInvoker;
                // 請求對應的Provider節(jié)點
                retryInvoker.invoke(invocation);
            } catch (Throwable e) {
                logger.error("Failed retry to invoke method " + invocation.getMethodName() + ", waiting again.", e);
                // 重試次數(shù)達到上限姨蟋，輸出警告日志
                if ((++retryTimes) >= retries) {
                    logger.error("Failed retry times exceed threshold (" + retries + "), We have to abandon, invocation->" + invocation);
                } else {
                    // 重試次數(shù)未達到上限，則重新添加定時任務立帖，等待重試
                    rePut(timeout);
                }
            }
        }

        private void rePut(Timeout timeout) {
            // 邊界檢查
            if (timeout == null) {
                return;
            }

            Timer timer = timeout.timer();
            // 檢查時間輪狀態(tài)眼溶、檢查定時任務狀態(tài)
            if (timer.isStop() || timeout.isCancelled()) {
                return;
            }
            // 重新添加定時任務
            timer.newTimeout(timeout.task(), tick, TimeUnit.SECONDS);
        }
    }       
}

FailfastClusterInvoker

FailfastCluster 的擴展名是 failfast，在其 doJoin() 方法中會創(chuàng)建 FailfastClusterInvoker 對象晓勇，具體實現(xiàn)如下：

public class FailfastCluster extends AbstractCluster {

    public final static String NAME = "failfast";

    @Override
    public <T> AbstractClusterInvoker<T> doJoin(Directory<T> directory) throws RpcException {
        return new FailfastClusterInvoker<>(directory);
    }

}

FailfastClusterInvoker 只會進行一次請求堂飞，請求失敗之后會立即拋出異常灌旧，這種策略適合非冪等的操作，具體實現(xiàn)如下：

public class FailfastClusterInvoker<T> extends AbstractClusterInvoker<T> {

    public FailfastClusterInvoker(Directory<T> directory) {
        super(directory);
    }

    @Override
    public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        checkInvokers(invokers, invocation);
        // 調(diào)用select()得到此次要調(diào)用的Invoker對象
        Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
        try {
            // 發(fā)起請求
            return invoker.invoke(invocation);
        } catch (Throwable e) {
            // 請求失敗绰筛，直接拋出異常
            if (e instanceof RpcException && ((RpcException) e).isBiz()) { // biz exception.
                throw (RpcException) e;
            }
            throw new RpcException(e instanceof RpcException ? ((RpcException) e).getCode() : 0,
                    "Failfast invoke providers " + invoker.getUrl() + " " + loadbalance.getClass().getSimpleName()
                            + " select from all providers " + invokers + " for service " + getInterface().getName()
                            + " method " + invocation.getMethodName() + " on consumer " + NetUtils.getLocalHost()
                            + " use dubbo version " + Version.getVersion()
                            + ", but no luck to perform the invocation. Last error is: " + e.getMessage(),
                    e.getCause() != null ? e.getCause() : e);
        }
    }
}

FailsafeClusterInvoker

FailsafeCluster 的擴展名是 failsafe枢泰，在其 doJoin() 方法中會創(chuàng)建 FailsafeClusterInvoker 對象，具體實現(xiàn)如下：

public class FailsafeCluster extends AbstractCluster {

    public final static String NAME = "failsafe";

    @Override
    public <T> AbstractClusterInvoker<T> doJoin(Directory<T> directory) throws RpcException {
        return new FailsafeClusterInvoker<>(directory);
    }

}

FailsafeClusterInvoker 只會進行一次請求铝噩，請求失敗之后會返回一個空結果衡蚂，具體實現(xiàn)如下：

public class FailsafeClusterInvoker<T> extends AbstractClusterInvoker<T> {
    private static final Logger logger = LoggerFactory.getLogger(FailsafeClusterInvoker.class);

    public FailsafeClusterInvoker(Directory<T> directory) {
        super(directory);
    }

    @Override
    public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        try {
            // 檢測Invoker集合是否為空
            checkInvokers(invokers, invocation);
            // 調(diào)用select()得到此次要調(diào)用的Invoker對象
            Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
            // 發(fā)起請求
            return invoker.invoke(invocation);
        } catch (Throwable e) {
            // 請求失敗之后，會打印一行日志并返回空結果
            logger.error("Failsafe ignore exception: " + e.getMessage(), e);
            return AsyncRpcResult.newDefaultAsyncResult(null, null, invocation); // ignore
        }
    }
}

ForkingClusterInvoker

ForkingCluster 的擴展名稱為 forking骏庸，在其 doJoin() 方法中毛甲，會創(chuàng)建一個 ForkingClusterInvoker 對象，具體實現(xiàn)如下：

public class ForkingCluster extends AbstractCluster {

    public final static String NAME = "forking";

    @Override
    public <T> AbstractClusterInvoker<T> doJoin(Directory<T> directory) throws RpcException {
        return new ForkingClusterInvoker<>(directory);
    }

}

ForkingClusterInvoker 中會維護一個線程池（executor 字段具被，通過 Executors.newCachedThreadPool() 方法創(chuàng)建的線程池）玻募，并發(fā)調(diào)用多個 Provider 節(jié)點，只要有一個 Provider 節(jié)點成功返回了結果一姿，F(xiàn)orkingClusterInvoker 的 doInvoke() 方法就會立即結束運行七咧。

ForkingClusterInvoker 主要是為了應對一些實時性要求較高的讀操作，因為沒有并發(fā)控制的多線程寫入叮叹，可能會導致數(shù)據(jù)不一致艾栋。

ForkingClusterInvoker.doInvoke() 方法首先從 Invoker 集合中選出指定個數(shù)（forks 參數(shù)決定）的 Invoker 對象，然后通過 executor 線程池并發(fā)調(diào)用這些 Invoker蛉顽，并將請求結果存儲在 ref 阻塞隊列中蝗砾，則當前線程會阻塞在 ref 隊列上，等待第一個請求結果返回蜂林。下面是 ForkingClusterInvoker 的具體實現(xiàn)：

public class ForkingClusterInvoker<T> extends AbstractClusterInvoker<T> {

    /**
     * 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 ExecutorService executor = Executors.newCachedThreadPool(
            new NamedInternalThreadFactory("forking-cluster-timer", true));

    public ForkingClusterInvoker(Directory<T> directory) {
        super(directory);
    }

    @Override
    @SuppressWarnings({"unchecked", "rawtypes"})
    public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        try {
            // 檢查Invoker集合是否為空
            checkInvokers(invokers, invocation);
            final List<Invoker<T>> selected;
            // 從URL中獲取forks參數(shù)，作為并發(fā)請求的上限拇泣，默認值為2
            final int forks = getUrl().getParameter(FORKS_KEY, DEFAULT_FORKS);
            final int timeout = getUrl().getParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT);
            if (forks <= 0 || forks >= invokers.size()) {
                // 如果forks為負數(shù)或是大于Invoker集合的長度噪叙，會直接并發(fā)調(diào)用全部Invoker
                selected = invokers;
            } else {
                // 按照forks指定的并發(fā)度，選擇此次并發(fā)調(diào)用的Invoker對象
                selected = new ArrayList<>(forks);
                while (selected.size() < forks) {
                    Invoker<T> invoker = select(loadbalance, invocation, invokers, selected);
                    if (!selected.contains(invoker)) {
                        //Avoid add the same invoker several times.
                        // 避免重復選擇
                        selected.add(invoker);
                    }
                }
            }
            RpcContext.getContext().setInvokers((List) selected);
            // 記錄失敗的請求個數(shù)
            final AtomicInteger count = new AtomicInteger();
            // 用于記錄請求結果
            final BlockingQueue<Object> ref = new LinkedBlockingQueue<>();
            // 遍歷 selected 列表
            for (final Invoker<T> invoker : selected) {
                // 為每個Invoker創(chuàng)建一個任務霉翔，并提交到線程池中
                executor.execute(() -> {
                    try {
                        // 發(fā)起請求
                        Result result = invoker.invoke(invocation);
                        // 將請求結果寫到ref隊列中
                        ref.offer(result);
                    } catch (Throwable e) {
                        int value = count.incrementAndGet();
                        // 如果失敗的請求個數(shù)超過了并發(fā)請求的個數(shù)睁蕾，則向ref隊列中寫入異常
                        if (value >= selected.size()) {
                            ref.offer(e);
                        }
                    }
                });
            }
            try {
                // 當前線程會阻塞等待任意一個請求結果的出現(xiàn)
                Object ret = ref.poll(timeout, TimeUnit.MILLISECONDS);
                // 如果結果類型為Throwable，則拋出異常
                if (ret instanceof Throwable) {
                    Throwable e = (Throwable) ret;
                    throw new RpcException(e instanceof RpcException ? ((RpcException) e).getCode() : 0, "Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e.getCause() != null ? e.getCause() : e);
                }
                // 返回結果
                return (Result) ret;
            } catch (InterruptedException e) {
                throw new RpcException("Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e);
            }
        } finally {
            // clear attachments which is binding to current thread.
            // 清除上下文信息
            RpcContext.getContext().clearAttachments();
        }
    }
}

BroadcastClusterInvoker

BroadcastCluster 這個 Cluster 實現(xiàn)類的擴展名為 broadcast债朵，在其 doJoin() 方法中創(chuàng)建的是 BroadcastClusterInvoker 類型的 Invoker 對象子眶，具體實現(xiàn)如下：

public class BroadcastCluster extends AbstractCluster {

    @Override
    public <T> AbstractClusterInvoker<T> doJoin(Directory<T> directory) throws RpcException {
        return new BroadcastClusterInvoker<>(directory);
    }

}

在 BroadcastClusterInvoker 中，會逐個調(diào)用每個 Provider 節(jié)點序芦，其中任意一個 Provider 節(jié)點報錯臭杰，都會在全部調(diào)用結束之后拋出異常。BroadcastClusterInvoker通常用于通知類的操作谚中，例如通知所有 Provider 節(jié)點更新本地緩存渴杆。

下面來看 BroadcastClusterInvoker 的具體實現(xiàn)：

public class BroadcastClusterInvoker<T> extends AbstractClusterInvoker<T> {

    private static final Logger logger = LoggerFactory.getLogger(BroadcastClusterInvoker.class);

    public BroadcastClusterInvoker(Directory<T> directory) {
        super(directory);
    }

    @Override
    @SuppressWarnings({"unchecked", "rawtypes"})
    public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        // 檢測Invoker集合是否為空
        checkInvokers(invokers, invocation);
        RpcContext.getContext().setInvokers((List) invokers);
        // 用于記錄失敗請求的相關異常信息
        RpcException exception = null;
        Result result = null;
        // 遍歷所有Invoker對象
        for (Invoker<T> invoker : invokers) {
            try {
                // 發(fā)起請求
                result = invoker.invoke(invocation);
            } catch (RpcException e) {
                exception = e;
                logger.warn(e.getMessage(), e);
            } catch (Throwable e) {
                exception = new RpcException(e.getMessage(), e);
                logger.warn(e.getMessage(), e);
            }
        }
        // 出現(xiàn)任何異常寥枝，都會在這里拋出
        if (exception != null) {
            throw exception;
        }
        return result;
    }

}

AvailableClusterInvoker

AvailableCluster 這個 Cluster 實現(xiàn)類的擴展名為 available，在其 join() 方法中創(chuàng)建的是 AvailableClusterInvoker 類型的 Invoker 對象磁奖，具體實現(xiàn)如下：

public class AvailableCluster implements Cluster {

    public static final String NAME = "available";

    @Override
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
        return new AvailableClusterInvoker<>(directory);
    }

}

在 AvailableClusterInvoker 的 doInvoke() 方法中囊拜，會遍歷整個 Invoker 集合，逐個調(diào)用對應的 Provider 節(jié)點比搭，當遇到第一個可用的 Provider 節(jié)點時冠跷，就嘗試訪問該 Provider 節(jié)點，成功則返回結果身诺；如果訪問失敗蜜托，則拋出異常終止遍歷。

下面是 AvailableClusterInvoker 的具體實現(xiàn)：

public class AvailableClusterInvoker<T> extends AbstractClusterInvoker<T> {

    public AvailableClusterInvoker(Directory<T> directory) {
        super(directory);
    }

    @Override
    public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        // 遍歷整個Invoker集合
        for (Invoker<T> invoker : invokers) {
            // 檢測該Invoker是否可用
            if (invoker.isAvailable()) {
                // 發(fā)起請求戚长，調(diào)用失敗時的異常會直接拋出
                return invoker.invoke(invocation);
            }
        }
        // 沒有找到可用的Invoker盗冷，也會拋出異常
        throw new RpcException("No provider available in " + invokers);
    }

}

MergeableClusterInvoker

MergeableCluster 這個 Cluster 實現(xiàn)類的擴展名為 mergeable，在其 doJoin() 方法中創(chuàng)建的是 MergeableClusterInvoker 類型的 Invoker 對象同廉，具體實現(xiàn)如下：

public class MergeableCluster extends AbstractCluster {

    public static final String NAME = "mergeable";

    @Override
    public <T> AbstractClusterInvoker<T> doJoin(Directory<T> directory) throws RpcException {
        return new MergeableClusterInvoker<T>(directory);
    }

}

MergeableClusterInvoker 會對多個 Provider 節(jié)點返回結果合并仪糖。如果請求的方法沒有配置 Merger 合并器（即沒有指定 merger 參數(shù)），則不會進行結果合并迫肖，而是直接將第一個可用的 Invoker 結果返回锅劝。下面來看 MergeableClusterInvoker 的具體實現(xiàn)：

public class MergeableClusterInvoker<T> extends AbstractClusterInvoker<T> {

    @Override
    protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        checkInvokers(invokers, invocation);
        String merger = getUrl().getMethodParameter(invocation.getMethodName(), MERGER_KEY);
        // 判斷要調(diào)用的目標方法是否有合并器，如果沒有蟆湖，則不會進行合并故爵，
        // 找到第一個可用的Invoker直接調(diào)用并返回結果     
        if (ConfigUtils.isEmpty(merger)) { // If a method doesn't have a merger, only invoke one Group
            for (final Invoker<T> invoker : invokers) {
                if (invoker.isAvailable()) {
                    try {
                        return invoker.invoke(invocation);
                    } catch (RpcException e) {
                        if (e.isNoInvokerAvailableAfterFilter()) {
                            log.debug("No available provider for service" + getUrl().getServiceKey() + " on group " + invoker.getUrl().getParameter(GROUP_KEY) + ", will continue to try another group.");
                        } else {
                            throw e;
                        }
                    }
                }
            }
            return invokers.iterator().next().invoke(invocation);
        }
        // 確定目標方法的返回值類型
        Class<?> returnType;
        try {
            returnType = getInterface().getMethod(
                    invocation.getMethodName(), invocation.getParameterTypes()).getReturnType();
        } catch (NoSuchMethodException e) {
            returnType = null;
        }
        // 調(diào)用每個Invoker對象(異步方式)，將請求結果記錄到results集合中
        Map<String, Result> results = new HashMap<>();
        for (final Invoker<T> invoker : invokers) {
            RpcInvocation subInvocation = new RpcInvocation(invocation, invoker);
            subInvocation.setAttachment(ASYNC_KEY, "true");
            results.put(invoker.getUrl().getServiceKey(), invoker.invoke(subInvocation));
        }

        Object result = null;

        List<Result> resultList = new ArrayList<Result>(results.size());
        // 等待結果返回
        for (Map.Entry<String, Result> entry : results.entrySet()) {
            Result asyncResult = entry.getValue();
            try {
                Result r = asyncResult.get();
                if (r.hasException()) {
                    log.error("Invoke " + getGroupDescFromServiceKey(entry.getKey()) +
                                    " failed: " + r.getException().getMessage(),
                            r.getException());
                } else {
                    resultList.add(r);
                }
            } catch (Exception e) {
                throw new RpcException("Failed to invoke service " + entry.getKey() + ": " + e.getMessage(), e);
            }
        }

        if (resultList.isEmpty()) {
            return AsyncRpcResult.newDefaultAsyncResult(invocation);
        } else if (resultList.size() == 1) {
            return resultList.iterator().next();
        }

        if (returnType == void.class) {
            return AsyncRpcResult.newDefaultAsyncResult(invocation);
        }
        // merger如果以"."開頭隅津，后面為方法名诬垂，這個方法名是遠程目標方法的返回類型中的方法
        // 得到每個Provider節(jié)點返回的結果對象之后，會遍歷每個返回對象伦仍，調(diào)用merger參數(shù)指定的方法
        if (merger.startsWith(".")) {
            merger = merger.substring(1);
            Method method;
            try {
                method = returnType.getMethod(merger, returnType);
            } catch (NoSuchMethodException e) {
                throw new RpcException("Can not merge result because missing method [ " + merger + " ] in class [ " +
                        returnType.getName() + " ]");
            }
            if (!Modifier.isPublic(method.getModifiers())) {
                method.setAccessible(true);
            }
            
            // resultList集合保存了所有的返回對象结窘，method是Method對象，也就是merger指定的方法
            // result是最后返回調(diào)用方的結果
            result = resultList.remove(0).getValue();
            try {
                if (method.getReturnType() != void.class
                        && method.getReturnType().isAssignableFrom(result.getClass())) {
                    for (Result r : resultList) {
                        result = method.invoke(result, r.getValue());
                    }
                } else {
                    for (Result r : resultList) {
                        // 反射調(diào)用
                        method.invoke(result, r.getValue());
                    }
                }
            } catch (Exception e) {
                throw new RpcException("Can not merge result: " + e.getMessage(), e);
            }
        } else {
            Merger resultMerger;
            if (ConfigUtils.isDefault(merger)) {
                // merger參數(shù)為true或者default充蓝，表示使用默認的Merger擴展實現(xiàn)完成合并
                resultMerger = MergerFactory.getMerger(returnType);
            } else {
                //merger參數(shù)指定了Merger的擴展名稱隧枫，則使用SPI查找對應的Merger擴展實現(xiàn)對象
                resultMerger = ExtensionLoader.getExtensionLoader(Merger.class).getExtension(merger);
            }
            if (resultMerger != null) {
                List<Object> rets = new ArrayList<Object>(resultList.size());
                for (Result r : resultList) {
                    rets.add(r.getValue());
                }
                // 執(zhí)行合并操作
                result = resultMerger.merge(
                        rets.toArray((Object[]) Array.newInstance(returnType, 0)));
            } else {
                throw new RpcException("There is no merger to merge result.");
            }
        }
        return AsyncRpcResult.newDefaultAsyncResult(result, invocation);
    }
}

ZoneAwareClusterInvoker

ZoneAwareCluster 這個 Cluster 實現(xiàn)類的擴展名為 zone-aware，在其 doJoin() 方法中創(chuàng)建的是 ZoneAwareClusterInvoker 類型的 Invoker 對象谓苟，具體實現(xiàn)如下：

public class ZoneAwareCluster extends AbstractCluster {

    public final static String NAME = "zone-aware";

    @Override
    protected <T> AbstractClusterInvoker<T> doJoin(Directory<T> directory) throws RpcException {
        return new ZoneAwareClusterInvoker<T>(directory);
    }

}

在 Dubbo 中使用多個注冊中心的架構如下圖所示：

雙注冊中心結構圖

Consumer 可以使用 ZoneAwareClusterInvoker 先在多個注冊中心之間進行選擇官脓，選定注冊中心之后，再選擇 Provider 節(jié)點涝焙，如下圖所示：

ZoneAwareClusterInvoker 在多注冊中心之間進行選擇的策略有以下四種：

• 1卑笨、找到preferred 屬性為 true 的注冊中心，它是優(yōu)先級最高的注冊中心仑撞，只有該中心無可用 Provider 節(jié)點時湾趾，才會回落到其他注冊中心芭商。

• 2、根據(jù)請求中的 zone key 做匹配搀缠，優(yōu)先派發(fā)到相同 zone 的注冊中心铛楣。

• 3、根據(jù)權重（也就是注冊中心配置的 weight 屬性）進行輪詢艺普。

• 4簸州、如果上面的策略都未命中，則選擇第一個可用的 Provider 節(jié)點歧譬。

下面來看 ZoneAwareClusterInvoker 的具體實現(xiàn)：

public class ZoneAwareClusterInvoker<T> extends AbstractClusterInvoker<T> {

    @Override
    @SuppressWarnings({"unchecked", "rawtypes"})
    public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        // First, pick the invoker (XXXClusterInvoker) that comes from the local registry, distinguish by a 'preferred' key.
        // 首先找到preferred屬性為true的注冊中心岸浑，它是優(yōu)先級最高的注冊中心，只有該中心無可用 Provider 節(jié)點時瑰步，才會回落到其他注冊中心
        for (Invoker<T> invoker : invokers) {
            ClusterInvoker<T> clusterInvoker = (ClusterInvoker<T>) invoker;
            if (clusterInvoker.isAvailable() && clusterInvoker.getRegistryUrl()
                    .getParameter(PREFERRED_KEY, false)) {
                return clusterInvoker.invoke(invocation);
            }
        }

        // providers in the registry with the same zone
        // 根據(jù)請求中的registry_zone做匹配矢洲，優(yōu)先派發(fā)到相同zone的注冊中心
        String zone = invocation.getAttachment(REGISTRY_ZONE);
        if (StringUtils.isNotEmpty(zone)) {
            for (Invoker<T> invoker : invokers) {
                ClusterInvoker<T> clusterInvoker = (ClusterInvoker<T>) invoker;
                if (clusterInvoker.isAvailable() && zone.equals(clusterInvoker.getRegistryUrl().getParameter(ZONE_KEY))) {
                    return clusterInvoker.invoke(invocation);
                }
            }
            String force = invocation.getAttachment(REGISTRY_ZONE_FORCE);
            if (StringUtils.isNotEmpty(force) && "true".equalsIgnoreCase(force)) {
                throw new IllegalStateException("No registry instance in zone or no available providers in the registry, zone: "
                        + zone
                        + ", registries: " + invokers.stream().map(invoker -> ((MockClusterInvoker<T>) invoker).getRegistryUrl().toString()).collect(Collectors.joining(",")));
            }
        }


        // load balance among all registries, with registry weight count in.
        // 根據(jù)權重（也就是注冊中心配置的weight屬性）進行輪詢
        Invoker<T> balancedInvoker = select(loadbalance, invocation, invokers, null);
        if (balancedInvoker.isAvailable()) {
            return balancedInvoker.invoke(invocation);
        }

        // If none of the invokers has a preferred signal or is picked by the loadbalancer, pick the first one available.
        // 選擇第一個可用的 Provider 節(jié)點
        for (Invoker<T> invoker : invokers) {
            ClusterInvoker<T> clusterInvoker = (ClusterInvoker<T>) invoker;
            if (clusterInvoker.isAvailable()) {
                return clusterInvoker.invoke(invocation);
            }
        }

        //if none available,just pick one
        return invokers.get(0).invoke(invocation);
    }

}

總結

本文重點介紹了 Dubbo 中 Cluster 接口的各個實現(xiàn)類的原理以及相關 Invoker 的實現(xiàn)原理。這里重點分析的 Cluster 實現(xiàn)有：Failover Cluster缩焦、Failback Cluster读虏、Failfast Cluster、Failsafe Cluster袁滥、Forking Cluster盖桥、Broadcast Cluster、Available Cluster 和 Mergeable Cluster题翻。除此之外揩徊，我們還分析了多注冊中心的 ZoneAware Cluster 實現(xiàn)。