Dubbo原始碼解析（三十五）叢集——cluster

阿新 • • 發佈：2020-06-24

叢集——cluster

目標：介紹dubbo中叢集容錯的幾種模式，介紹dubbo-cluster下support包的原始碼。

前言

叢集容錯還是很好理解的，就是當你呼叫失敗的時候所作出的措施。先來看看有哪些模式：

圖有點小，見諒，不過可以眯著眼睛看稍微能看出來一點，每一個Cluster實現類都對應著一個invoker，因為這個模式啟用的時間點就是在呼叫的時候，而我在之前的文章裡面講過，invoker貫穿來整個服務的呼叫。不過這裡除了呼叫失敗的一些模式外，還有幾個特別的模式，他們應該說成是失敗的措施，而已呼叫的方式。

Failsafe Cluster：失敗安全，出現異常時，直接忽略。失敗安全就是當呼叫過程中出現異常時，FailsafeClusterInvoker 僅會列印異常，而不會丟擲異常。適用於寫入審計日誌等操作

Failover Cluster：失敗自動切換，當調用出現失敗的時候，會自動切換叢集中其他伺服器，來獲得invoker重試，通常用於讀操作，但重試會帶來更長延遲。一般都會設定重試次數。
Failfast Cluster：只會進行一次呼叫，失敗後立即丟擲異常。適用於冪等操作，比如新增記錄。
Failback Cluster：失敗自動恢復，在呼叫失敗後，返回一個空結果給服務提供者。並通過定時任務對失敗的呼叫記錄並且重傳，適合執行訊息通知等操作。
Forking Cluster：會線上程池中執行多個執行緒，來呼叫多個伺服器，只要一個成功即返回。通常用於實時性要求較高的讀操作，但需要浪費更多服務資源。一般會設定最大並行數。

Available Cluster：呼叫第一個可用的伺服器，僅僅應用於多註冊中心。
Broadcast Cluster：廣播呼叫所有提供者，逐個呼叫，在迴圈呼叫結束後，只要任意一臺報錯就報錯。通常用於通知所有提供者更新快取或日誌等本地資源資訊
Mergeable Cluster：該部分在分組聚合講述。
MockClusterWrapper：該部分在本地偽裝講述。

原始碼分析

（一）AbstractClusterInvoker

該類實現了Invoker介面，是叢集Invoker的抽象類。

1.屬性

private static final Logger logger = LoggerFactory
        .getLogger(AbstractClusterInvoker.class);
/**
 * 目錄，包含多個invoker
 */ 

protected final Directory<T> directory;

/**
 * 是否需要核對可用
 */
protected final boolean availablecheck;

/**
 * 是否銷燬
 */
private AtomicBoolean destroyed = new AtomicBoolean(false);

/**
 * 粘滯連線的Invoker
 */
private volatile Invoker<T> stickyInvoker = null;
複製程式碼

2.select

protected Invoker<T> select(LoadBalance loadbalance,Invocation invocation,List<Invoker<T>> invokers,List<Invoker<T>> selected) throws RpcException {
    // 如果invokers為空，則返回null
    if (invokers == null || invokers.isEmpty())
        return null;
    // 獲得方法名
    String methodName = invocation == null ? "" : invocation.getMethodName();

    // 是否啟動了粘滯連線
    boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName,Constants.CLUSTER_STICKY_KEY,Constants.DEFAULT_CLUSTER_STICKY);
    {
        //ignore overloaded method
        // 如果上一次粘滯連線的呼叫不在可選的提供者列合內，則直接設定為空
        if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
            stickyInvoker = null;
        }
        //ignore concurrency problem
        // stickyInvoker不為null,並且沒在已選列表中，返回上次的服務提供者stickyInvoker，但之前強制校驗可達性。
        // 由於stickyInvoker不能包含在selected列表中，通過程式碼看，可以得知forking和failover叢集策略，用不了sticky屬性
        if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
            if (availablecheck && stickyInvoker.isAvailable()) {
                return stickyInvoker;
            }
        }
    }
    // 利用負載均衡選一個提供者
    Invoker<T> invoker = doSelect(loadbalance,invocation,invokers,selected);

    // 如果啟動粘滯連線，則記錄這一次的呼叫
    if (sticky) {
        stickyInvoker = invoker;
    }
    return invoker;
}
複製程式碼

該方法實現了使用負載均衡策略選擇一個呼叫者。首先，使用loadbalance選擇一個呼叫者。如果此呼叫者位於先前選擇的列表中，或者如果此呼叫者不可用，則重新選擇，否則返回第一個選定的呼叫者。重新選擇，重選的驗證規則：選擇>可用。這條規則可以保證所選的呼叫者最少有機會成為之前選擇的列表中的一個，也是保證這個呼叫程式可用。

3.doSelect

private Invoker<T> doSelect(LoadBalance loadbalance,List<Invoker<T>> selected) throws RpcException {
    if (invokers == null || invokers.isEmpty())
        return null;
    // 如果只有一個 ，就直接返回這個
    if (invokers.size() == 1)
        return invokers.get(0);
    // 如果沒有指定用哪個負載均衡策略，則預設用隨機負載均衡策略
    if (loadbalance == null) {
        loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
    }
    // 呼叫負載均衡選擇
    Invoker<T> invoker = loadbalance.select(invokers,getUrl(),invocation);

    //If the `invoker` is in the  `selected` or invoker is unavailable && availablecheck is true,reselect.
    // 如果選擇的提供者，已在selected中或者不可用則重新選擇
    if ((selected != null && selected.contains(invoker))
            || (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
        try {
            // 重新選擇
            Invoker<T> rinvoker = reselect(loadbalance,selected,availablecheck);
            if (rinvoker != null) {
                invoker = rinvoker;
            } else {
                //Check the index of current selected invoker,if it's not the last one,choose the one at index+1.
                // 如果重新選擇失敗，看下第一次選的位置，如果不是最後，選+1位置.
                int index = invokers.indexOf(invoker);
                try {
                    //Avoid collision
                    // 最後再避免選擇到同一個invoker
                    invoker = index < invokers.size() - 1 ? invokers.get(index + 1) : invokers.get(0);
                } catch (Exception e) {
                    logger.warn(e.getMessage() + " may because invokers list dynamic change,ignore.",e);
                }
            }
        } catch (Throwable t) {
            logger.error("cluster reselect fail reason is :" + t.getMessage() + " if can not solve,you can set cluster.availablecheck=false in url",t);
        }
    }
    return invoker;
}
複製程式碼

該方法是用負載均衡選擇一個invoker的主要邏輯。

4.reselect

private Invoker<T> reselect(LoadBalance loadbalance,List<Invoker<T>> selected,boolean availablecheck)
        throws RpcException {

    //Allocating one in advance,this list is certain to be used.
    //預先分配一個重選列表，這個列表是一定會用到的.
    List<Invoker<T>> reselectInvokers = new ArrayList<Invoker<T>>(invokers.size() > 1 ? (invokers.size() - 1) : invokers.size());

    //First,try picking a invoker not in `selected`.
    //先從非select中選
    //把不包含在selected中的提供者，放入重選列表reselectInvokers，讓負載均衡器選擇
    if (availablecheck) { // invoker.isAvailable() should be checked
        for (Invoker<T> invoker : invokers) {
            if (invoker.isAvailable()) {
                if (selected == null || !selected.contains(invoker)) {
                    reselectInvokers.add(invoker);
                }
            }
        }
        // 在重選列表中用負載均衡器選擇
        if (!reselectInvokers.isEmpty()) {
            return loadbalance.select(reselectInvokers,invocation);
        }
    } else { // do not check invoker.isAvailable()
        // 不核對服務是否可以，把不包含在selected中的提供者，放入重選列表reselectInvokers，讓負載均衡器選擇
        for (Invoker<T> invoker : invokers) {
            if (selected == null || !selected.contains(invoker)) {
                reselectInvokers.add(invoker);
            }
        }
        if (!reselectInvokers.isEmpty()) {
            return loadbalance.select(reselectInvokers,invocation);
        }
    }
    // Just pick an available invoker using loadbalance policy
    {
        // 如果非selected的列表中沒有選擇到，則從selected中選擇
        if (selected != null) {
            for (Invoker<T> invoker : selected) {
                if ((invoker.isAvailable()) // available first
                        && !reselectInvokers.contains(invoker)) {
                    reselectInvokers.add(invoker);
                }
            }
        }
        if (!reselectInvokers.isEmpty()) {
            return loadbalance.select(reselectInvokers,invocation);
        }
    }
    return null;
}
複製程式碼

該方法是是重新選擇的邏輯實現。

5.invoke

@Override
public Result invoke(final Invocation invocation) throws RpcException {
    // 核對是否已經銷燬
    checkWhetherDestroyed();
    LoadBalance loadbalance = null;

    // binding attachments into invocation.
    // 獲得上下文的附加值
    Map<String,String> contextAttachments = RpcContext.getContext().getAttachments();
    // 把附加值放入到會話域中
    if (contextAttachments != null && contextAttachments.size() != 0) {
        ((RpcInvocation) invocation).addAttachments(contextAttachments);
    }

    // 生成服務提供者集合
    List<Invoker<T>> invokers = list(invocation);
    if (invokers != null && !invokers.isEmpty()) {
        // 獲得負載均衡器
        loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
                .getMethodParameter(RpcUtils.getMethodName(invocation),Constants.LOADBALANCE_KEY,Constants.DEFAULT_LOADBALANCE));
    }
    RpcUtils.attachInvocationIdIfAsync(getUrl(),invocation);
    return doInvoke(invocation,loadbalance);
}
複製程式碼

該方法是invoker介面必備的方法，呼叫鏈的邏輯，不過主要的邏輯在doInvoke方法中，該方法是該類的抽象方法，讓子類只關注doInvoke方法。

6.list

protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
    // 把會話域中的invoker加入集合
    List<Invoker<T>> invokers = directory.list(invocation);
    return invokers;
}
複製程式碼

該方法是呼叫了directory的list方法，從會話域中獲得所有的Invoker集合。關於directory我會在後續文章講解。

（二）AvailableCluster

public class AvailableCluster implements Cluster {

    public static final String NAME = "available";

    @Override
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException {

        // 建立一個AbstractClusterInvoker
        return new AbstractClusterInvoker<T>(directory) {
            @Override
            public Result doInvoke(Invocation invocation,LoadBalance loadbalance) throws RpcException {
                // 遍歷所有的involer，只要有一個可用就直接呼叫。
                for (Invoker<T> invoker : invokers) {
                    if (invoker.isAvailable()) {
                        return invoker.invoke(invocation);
                    }
                }
                throw new RpcException("No provider available in " + invokers);
            }
        };

    }

}
複製程式碼

Available Cluster我在上面已經講過了，只要找到一個可用的，則直接呼叫。

（三）BroadcastCluster

public class BroadcastCluster implements Cluster {

    @Override
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
        // 建立一個BroadcastClusterInvoker
        return new BroadcastClusterInvoker<T>(directory);
    }

}
複製程式碼

關鍵實現在於BroadcastClusterInvoker。

（四）BroadcastClusterInvoker

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,LoadBalance loadbalance) throws RpcException {
        // 檢測invokers是否為空
        checkInvokers(invokers,invocation);
        // 把invokers放到上下文
        RpcContext.getContext().setInvokers((List) invokers);
        RpcException exception = null;
        Result result = null;
        // 遍歷invokers，逐個呼叫，在迴圈呼叫結束後，只要任意一臺報錯就報錯
        for (Invoker<T> invoker : invokers) {
            try {
                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);
            }
        }
        if (exception != null) {
            throw exception;
        }
        return result;
    }

}
複製程式碼

（五）ForkingCluster

public class ForkingCluster implements Cluster {

    public final static String NAME = "forking";

    @Override
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
        // 建立ForkingClusterInvoker
        return new ForkingClusterInvoker<T>(directory);
    }

}
複製程式碼

（六）ForkingClusterInvoker

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

    /**
     * 執行緒池
     * Use {@link NamedInternalThreadFactory} to produce {@link com.alibaba.dubbo.common.threadlocal.InternalThread}
     * which with the use of {@link com.alibaba.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",LoadBalance loadbalance) throws RpcException {
        try {
            // 檢測invokers是否為空
            checkInvokers(invokers,invocation);
            final List<Invoker<T>> selected;
            // 獲取 forks 配置
            final int forks = getUrl().getParameter(Constants.FORKS_KEY,Constants.DEFAULT_FORKS);
            // 獲取超時配置
            final int timeout = getUrl().getParameter(Constants.TIMEOUT_KEY,Constants.DEFAULT_TIMEOUT);
            // 如果 forks 配置不合理，則直接將 invokers 賦值給 selected
            if (forks <= 0 || forks >= invokers.size()) {
                selected = invokers;
            } else {
                selected = new ArrayList<Invoker<T>>();
                // 迴圈選出 forks 個 Invoker，並新增到 selected 中
                for (int i = 0; i < forks; i++) {
                    // TODO. Add some comment here,refer chinese version for more details.
                    // 選擇 Invoker
                    Invoker<T> invoker = select(loadbalance,selected);
                    if (!selected.contains(invoker)) {//Avoid add the same invoker several times.
                        // 加入到selected集合
                        selected.add(invoker);
                    }
                }
            }
            // 放入上下文
            RpcContext.getContext().setInvokers((List) selected);
            final AtomicInteger count = new AtomicInteger();
            final BlockingQueue<Object> ref = new LinkedBlockingQueue<Object>();
            // 遍歷 selected 列表
            for (final Invoker<T> invoker : selected) {
                // 為每個 Invoker 建立一個執行執行緒
                executor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            // 進行遠端呼叫
                            Result result = invoker.invoke(invocation);
                            // 將結果存到阻塞佇列中
                            ref.offer(result);
                        } catch (Throwable e) {
                            // 僅在 value 大於等於 selected.size() 時，才將異常物件
                            // 為了防止異常現象覆蓋正常的結果
                            int value = count.incrementAndGet();
                            if (value >= selected.size()) {
                                // 將異常物件存入到阻塞佇列中
                                ref.offer(e);
                            }
                        }
                    }
                });
            }
            try {
                // 從阻塞佇列中取出遠端呼叫結果
                Object ret = ref.poll(timeout,TimeUnit.MILLISECONDS);
                // 如果是異常，則丟擲
                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 + ",e);
            }
        } finally {
            // clear attachments which is binding to current thread.
            RpcContext.getContext().clearAttachments();
        }
    }
}
複製程式碼

（七）FailbackCluster

public class FailbackCluster implements Cluster {

    public final static String NAME = "failback";

    @Override
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
        // 建立一個FailbackClusterInvoker
        return new FailbackClusterInvoker<T>(directory);
    }

}
複製程式碼

（八）FailbackClusterInvoker

1.屬性

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

// 重試間隔
private static final long RETRY_FAILED_PERIOD = 5 * 1000;

/**
 * 定時器
 * Use {@link NamedInternalThreadFactory} to produce {@link com.alibaba.dubbo.common.threadlocal.InternalThread}
 * which with the use of {@link com.alibaba.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<?>>();
/**
 * future
 */
private volatile ScheduledFuture<?> retryFuture;
複製程式碼

2.doInvoke

@Override
protected Result doInvoke(Invocation invocation,LoadBalance loadbalance) throws RpcException {
    try {
        // 檢測invokers是否為空
        checkInvokers(invokers,invocation);
        // 選擇出invoker
        Invoker<T> invoker = select(loadbalance,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
    }
}
複製程式碼

該方法是選擇invoker呼叫的邏輯，在丟擲異常的時候，做了失敗重試的機制，主要實現在addFailed。

3.addFailed

private void addFailed(Invocation invocation,AbstractClusterInvoker<?> router) {
    if (retryFuture == null) {
        // 鎖住
        synchronized (this) {
            if (retryFuture == null) {
                // 建立定時任務，每隔5秒執行一次
                retryFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {

                    @Override
                    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,TimeUnit.MILLISECONDS);
            }
        }
    }
    // 新增 invocation 和 invoker 到 failed 中
    failed.put(invocation,router);
}
複製程式碼

該方法做的事建立了定時器，然後把失敗的呼叫放入到集合中。

4.retryFailed

void retryFailed() {
    // 如果失敗佇列為0，返回
    if (failed.size() == 0) {
        return;
    }
    // 遍歷失敗佇列
    for (Map.Entry<Invocation,AbstractClusterInvoker<?>> entry : new HashMap<Invocation,AbstractClusterInvoker<?>>(
            failed).entrySet()) {
        // 獲得會話域
        Invocation invocation = entry.getKey();
        // 獲得invoker
        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);
        }
    }
}
複製程式碼

這個方法是呼叫失敗的invoker重新呼叫的機制。

（九）FailfastCluster

public class FailfastCluster implements Cluster {

    public final static String NAME = "failfast";

    @Override
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
        // 建立FailfastClusterInvoker
        return new FailfastClusterInvoker<T>(directory);
    }

}

複製程式碼

（十）FailfastClusterInvoker

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

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

    @Override
    public Result doInvoke(Invocation invocation,invocation);
        // 選擇一個invoker
        Invoker<T> invoker = select(loadbalance,null);
        try {
            // 呼叫
            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() + ",e.getCause() != null ? e.getCause() : e);
        }
    }
}

複製程式碼

邏輯比較簡單，呼叫丟擲異常就直接丟擲。

（十一）FailoverCluster

public class FailoverCluster implements Cluster {

    public final static String NAME = "failover";

    @Override
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
        // 建立FailoverClusterInvoker
        return new FailoverClusterInvoker<T>(directory);
    }

}

複製程式碼

（十二）FailoverClusterInvoker

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

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

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

    @Override
    @SuppressWarnings({"unchecked","rawtypes"})
    public Result doInvoke(Invocation invocation,final List<Invoker<T>> invokers,LoadBalance loadbalance) throws RpcException {
        // 複製一個invoker集合
        List<Invoker<T>> copyinvokers = invokers;
        // 檢測是否為空
        checkInvokers(copyinvokers,invocation);
        // 獲取重試次數
        int len = getUrl().getMethodParameter(invocation.getMethodName(),Constants.RETRIES_KEY,Constants.DEFAULT_RETRIES) + 1;
        if (len <= 0) {
            len = 1;
        }
        // retry loop.
        // 記錄最後一個異常
        RpcException le = null; // last exception.
        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++) {
            //Reselect before retry to avoid a change of candidate `invokers`.
            //NOTE: if `invokers` changed,then `invoked` also lose accuracy.
            // 在進行重試前重新列舉 Invoker，這樣做的好處是，如果某個服務掛了，
            // 通過呼叫 list 可得到最新可用的 Invoker 列表
            if (i > 0) {
                checkWhetherDestroyed();
                copyinvokers = list(invocation);
                // check again
                // 檢測copyinvokers 是否為空
                checkInvokers(copyinvokers,invocation);
            }
            // 通過負載均衡選擇invoker
            Invoker<T> invoker = select(loadbalance,copyinvokers,invoked);
            // 新增到 invoker 到 invoked 列表中
            invoked.add(invoker);
            // 設定 invoked 到 RPC 上下文中
            RpcContext.getContext().setInvokers((List) invoked);
            try {
                // 呼叫目標 Invoker 的 invoke 方法
                Result result = invoker.invoke(invocation);
                if (le != null && logger.isWarnEnabled()) {
                    logger.warn("Although retry the method " + invocation.getMethodName()
                            + " 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 {
                providers.add(invoker.getUrl().getAddress());
            }
        }
        // 若重試失敗，則丟擲異常
        throw new RpcException(le != null ? le.getCode() : 0,"Failed to invoke the method "
                + invocation.getMethodName() + " 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 != null ? le.getMessage() : ""),le != null && le.getCause() != null ? le.getCause() : le);
    }

}

複製程式碼

該類實現了失敗重試的容錯策略，當呼叫失敗的時候，記錄下異常，然後迴圈呼叫下一個選擇出來的invoker，直到重試次數用完，丟擲最後一次的異常。

（十三）FailsafeCluster

public class FailsafeCluster implements Cluster {

    public final static String NAME = "failsafe";

    @Override
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
        // 建立FailsafeClusterInvoker
        return new FailsafeClusterInvoker<T>(directory);
    }

}

複製程式碼

（十四）FailsafeClusterInvoker

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,invocation);
            // 選擇一個invoker
            Invoker<T> invoker = select(loadbalance,null);
            // 呼叫
            return invoker.invoke(invocation);
        } catch (Throwable e) {
            // 如果失敗列印異常，返回一個空結果
            logger.error("Failsafe ignore exception: " + e.getMessage(),e);
            return new RpcResult(); // ignore
        }
    }
}

複製程式碼

邏輯比較簡單，就是不丟擲異常，只是列印異常。

後記

該部分相關的原始碼解析地址：github.com/CrazyHZM/in…

該文章講解了叢集中關於cluster實現的部分，講了幾種呼叫方式和容錯策略。接下來我將開始對叢集模組關於配置規則部分進行講解。

Dubbo原始碼解析（三十五）叢集——cluster

叢集——cluster

前言

原始碼分析

（一）AbstractClusterInvoker

1.屬性

2.select

3.doSelect

4.reselect

5.invoke

6.list

（二）AvailableCluster

（三）BroadcastCluster

（四）BroadcastClusterInvoker

（五）ForkingCluster

（六）ForkingClusterInvoker

（七）FailbackCluster

（八）FailbackClusterInvoker

1.屬性

2.doInvoke

3.addFailed

4.retryFailed

（九）FailfastCluster

（十）FailfastClusterInvoker

（十一）FailoverCluster

（十二）FailoverClusterInvoker

（十三）FailsafeCluster

（十四）FailsafeClusterInvoker

後記

相關推薦