dubbo原始碼解析五 --- 叢集容錯架構設計與原理分析
阿新 • • 發佈：2019-01-24
歡迎來我的 Star Followers 後期後繼續更新Dubbo別的文章
下面是個人部落格地址，頁面比部落格園美觀一些其他都是一樣的
目錄
- 面試中叢集容錯的經常的問題
- Dubbo 官方文件關於叢集容錯的介紹
- Dubbo叢集容錯的架構分析
- Dubbo叢集容錯原始碼解析
面試中叢集容錯的經常的問題
- 什麼是叢集容錯
- Dubbo的叢集容錯知道嗎
- Dubbo 叢集容錯是如何配置的
- 叢集容錯如何實現的
- Dubbo 叢集容錯介紹下
- 介紹下 幾種叢集容錯方式，分析下其優缺點
- 你來設計一個容錯演算法，你會怎樣的設計
Dubbo 官方文件關於叢集容錯的介紹
在叢集呼叫失敗時，Dubbo 提供了多種容錯方案，預設為 failover 重試。
各節點關係：
- 這裡的
Invoker是Provider的一個可呼叫Service的抽象，Invoker封裝了Provider地址及Service介面資訊 Directory代表多個Invoker，可以把它看成List<Invoker>，但與List不同的是，它的值可能是動態變化的，比如註冊中心推送變更Cluster將Directory中的多個Invoker偽裝成一個Invoker，對上層透明，偽裝過程包含了容錯邏輯，呼叫失敗後，重試另一個Router負責從多個Invoker中按路由規則選出子集，比如讀寫分離，應用隔離等LoadBalance負責從多個Invoker中選出具體的一個用於本次呼叫，選的過程包含了負載均衡演算法，呼叫失敗後，需要重選
叢集容錯模式
可以自行擴充套件叢集容錯策略，參見：叢集擴充套件
失敗自動切換，當出現失敗，重試其它伺服器 [1]。通常用於讀操作，但重試會帶來更長延遲。可通過 retries="2" 來設定重試次數(不含第一次)。
重試次數配置如下：
<dubbo:service retries="2" />或
<dubbo:reference retries="2" />或
<dubbo:reference> <dubbo:method name="findFoo" retries="2" /> </dubbo:reference>
快速失敗，只發起一次呼叫，失敗立即報錯。通常用於非冪等性的寫操作，比如新增記錄。
失敗安全，出現異常時，直接忽略。通常用於寫入審計日誌等操作。
失敗自動恢復，後臺記錄失敗請求，定時重發。通常用於訊息通知操作。
並行呼叫多個伺服器，只要一個成功即返回。通常用於實時性要求較高的讀操作，但需要浪費更多服務資源。可通過 forks="2" 來設定最大並行數。
廣播呼叫所有提供者，逐個呼叫，任意一臺報錯則報錯 [2]。通常用於通知所有提供者更新快取或日誌等本地資源資訊。
叢集模式配置
按照以下示例在服務提供方和消費方配置叢集模式
<dubbo:service cluster="failsafe" />或
<dubbo:reference cluster="failsafe" />Dubbo叢集容錯的架構分析
通過官網上這張圖我們能大致的瞭解到一個請求過來，在叢集中的呼叫過程。那麼我們就根據這個呼叫過程來進行分析吧。
整個在呼叫的過程中 這三個關鍵詞接下來會貫穿全文,他們就是Directory,Router,LoadBalance
我們只要牢牢的抓住這幾個關鍵字就能貫穿整個呼叫鏈
先看下時序圖，來看下呼叫的過程
最初我們一個方法呼叫
我們使用的是官方的dubbo-demo的dubbo-demo-consumer
public static void main(String[] args) {
DemoService demoService = (DemoService) context.getBean("demoService"); // get remote service proxy
String hello = demoService.sayHello("world"); // call remote method
System.out.println(hello); // get result
}public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
String methodName = method.getName();
Class<?>[] parameterTypes = method.getParameterTypes();
if (method.getDeclaringClass() == Object.class) {
return method.invoke(invoker, args);
}
if ("toString".equals(methodName) && parameterTypes.length == 0) {
return invoker.toString();
}
if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
return invoker.hashCode();
}
if ("equals".equals(methodName) && parameterTypes.length == 1) {
return invoker.equals(args[0]);
}
RpcInvocation invocation;
if (RpcUtils.hasGeneratedFuture(method)) {
Class<?> clazz = method.getDeclaringClass();
String syncMethodName = methodName.substring(0, methodName.length() - Constants.ASYNC_SUFFIX.length());
Method syncMethod = clazz.getMethod(syncMethodName, method.getParameterTypes());
invocation = new RpcInvocation(syncMethod, args);
invocation.setAttachment(Constants.FUTURE_GENERATED_KEY, "true");
invocation.setAttachment(Constants.ASYNC_KEY, "true");
} else {
invocation = new RpcInvocation(method, args);
if (RpcUtils.hasFutureReturnType(method)) {
invocation.setAttachment(Constants.FUTURE_RETURNTYPE_KEY, "true");
invocation.setAttachment(Constants.ASYNC_KEY, "true");
}
}
//這裡使用的是動態代理的方式獲取到指定的代理類
// <1>
return invoker.invoke(invocation).recreate();
}public Result invoke(Invocation invocation) throws RpcException {
Result result = null;
// 獲得 “mock” 配置項，有多種配置方式
String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), Constants.MOCK_KEY, Boolean.FALSE.toString()).trim();
//【第一種】無 mock
if (value.length() == 0 || value.equalsIgnoreCase("false")) {
//no mock
// 呼叫原 Invoker ，發起 RPC 呼叫
// 呼叫 invoker方法，進入到叢集也就是CLuster類中
//<2>
result = this.invoker.invoke(invocation);
//【第二種】強制服務降級
} else if (value.startsWith("force")) {
if (logger.isWarnEnabled()) {
logger.warn("force-mock: " + invocation.getMethodName() + " force-mock enabled , url : " + directory.getUrl());
}
//force:direct mock
// 直接呼叫 Mock Invoker ，執行本地 Mock 邏輯
result = doMockInvoke(invocation, null);
} else {
//fail-mock
try {
// 【第三種】失敗服務降級
result = this.invoker.invoke(invocation);
} catch (RpcException e) {
// 業務性異常，直接丟擲
if (e.isBiz()) {
throw e;
} else {
if (logger.isWarnEnabled()) {
logger.warn("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " + directory.getUrl(), e);
}
// 失敗後，呼叫 Mock Invoker ，執行本地 Mock 邏輯
result = doMockInvoke(invocation, e);
}
}
}
return result;
}2 進入到 invoke就要開始進入到叢集,也就是Cluster
/**
* 呼叫服務提供者
* @param invocation
* @return
* @throws RpcException
*/
@Override
public Result invoke(final Invocation invocation) throws RpcException {
// 校驗是否銷燬
checkWhetherDestroyed();
//TODO
// binding attachments into invocation.
Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
if (contextAttachments != null && contextAttachments.size() != 0) {
((RpcInvocation) invocation).addAttachments(contextAttachments);
}
// 獲得所有服務提供者 Invoker 集合
// <下面的list方法>
List<Invoker<T>> invokers = list(invocation);
// 獲得 LoadBalance 物件
LoadBalance loadbalance = initLoadBalance(invokers, invocation);
// 設定呼叫編號，若是非同步呼叫
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
// 執行呼叫
return doInvoke(invocation, invokers, loadbalance);
}/**
* 獲得所有服務提供者 Invoker 集合
* @param invocation
* @return
* @throws RpcException
*/
protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
// 通過directory 進入到 AbstractDirectory 中選擇 directory
//<3 進入3 裡面>
return directory.list(invocation);
}/**
* 獲得所有服務 Invoker 集合
* @param invocation
* @return Invoker 集合
* @throws RpcException
*/
@Override
public List<Invoker<T>> list(Invocation invocation) throws RpcException {
//當銷燬時丟擲異常
if (destroyed) {
throw new RpcException("Directory already destroyed .url: " + getUrl());
}
// 獲得所有 Invoker 集合
// <4 RegistryDirectory 選擇 invoker>
List<Invoker<T>> invokers = doList(invocation);
//根據路由規則，篩選Invoker集合
List<Router> localRouters = this.routers; // local reference 本地引用，避免併發問題
if (localRouters != null && !localRouters.isEmpty()) {
for (Router router : localRouters) {
try {
if (router.getUrl() == null || router.getUrl().getParameter(Constants.RUNTIME_KEY, false)) {
invokers = router.route(invokers, getConsumerUrl(), invocation);
}
} catch (Throwable t) {
logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
}
}
}
//< 6 獲取 router 即將進入 MockInvokersSelector 類中>
return invokers;
}/**
* 獲得對應的 Invoker 集合。
* @param invocation
* @return
*/
@Override
public List<Invoker<T>> doList(Invocation invocation) {
if (forbidden) {
// 1. No service provider 2. Service providers are disabled
throw new RpcException(RpcException.FORBIDDEN_EXCEPTION,
"No provider available from registry " + getUrl().getAddress() + " for service " + getConsumerUrl().getServiceKey() + " on consumer " + NetUtils.getLocalHost()
+ " use dubbo version " + Version.getVersion() + ", please check status of providers(disabled, not registered or in blacklist).");
}
List<Invoker<T>> invokers = null;
//從methodInvokerMap中取出invokers
Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap; // local reference
// 獲得 Invoker 集合
if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {
// 獲得方法名、方法引數
String methodName = RpcUtils.getMethodName(invocation);
Object[] args = RpcUtils.getArguments(invocation);
// 【第一】可根據第一個引數列舉路由
if (args != null && args.length > 0 && args[0] != null
&& (args[0] instanceof String || args[0].getClass().isEnum())) {
invokers = localMethodInvokerMap.get(methodName + "." + args[0]); // The routing can be enumerated according to the first parameter
}
// 【第二】根據方法名獲得 Invoker 集合
if (invokers == null) {
invokers = localMethodInvokerMap.get(methodName);
}
// 【第三】使用全量 Invoker 集合。例如，`#$echo(name)` ，回聲方法
if (invokers == null) {
invokers = localMethodInvokerMap.get(Constants.ANY_VALUE);
}
// 【第四】使用 `methodInvokerMap` 第一個 Invoker 集合。防禦性程式設計。
if (invokers == null) {
Iterator<List<Invoker<T>>> iterator = localMethodInvokerMap.values().iterator();
if (iterator.hasNext()) {
invokers = iterator.next();
}
}
}
return invokers == null ? new ArrayList<Invoker<T>>(0) : invokers;
}/**
* ，根據 "invocation.need.mock" 路由匹配對應型別的 Invoker 集合：
* @param invokers Invoker 集合
* @param url refer url
* @param invocation
* @param <T>
* @return
* @throws RpcException
*/
@Override
public <T> List<Invoker<T>> route(final List<Invoker<T>> invokers,
URL url, final Invocation invocation) throws RpcException {
// 獲得普通 Invoker 集合
if (invocation.getAttachments() == null) {
//<7> 拿到能正常執行的invokers,並將其返回.也就是序號7
return getNormalInvokers(invokers);
} else {
// 獲得 "invocation.need.mock" 配置項
String value = invocation.getAttachments().get(Constants.INVOCATION_NEED_MOCK);
// 獲得普通 Invoker 集合
if (value == null)
return getNormalInvokers(invokers);
// 獲得 MockInvoker 集合
else if (Boolean.TRUE.toString().equalsIgnoreCase(value)) {
return getMockedInvokers(invokers);
}
}
// 其它，不匹配，直接返回 `invokers` 集合
return invokers;
}
<7> 拿到能正常執行的invokers,並將其返回
/**
* 獲得普通 Invoker 集合
* @param invokers
* @param <T>
* @return
*/
private <T> List<Invoker<T>> getNormalInvokers(final List<Invoker<T>> invokers) {
// 不包含 MockInvoker 的情況下，直接返回 `invokers` 集合
if (!hasMockProviders(invokers)) {
return invokers;
} else {
// 若包含 MockInvoker 的情況下，過濾掉 MockInvoker ，建立普通 Invoker 集合
List<Invoker<T>> sInvokers = new ArrayList<Invoker<T>>(invokers.size());
for (Invoker<T> invoker : invokers) {
if (!invoker.getUrl().getProtocol().equals(Constants.MOCK_PROTOCOL)) {
sInvokers.add(invoker);
}
}
return sInvokers;
}
}8 拿到 invoker 返回到 AbstractClusterInvoker這個類
對於上面的這些步驟，主要用於做兩件事
- 在
Directory中找出本次叢集中的全部invokers- 在
Router中,將上一步的全部invokers挑選出能正常執行的invokers在 時序圖的序號5和序號7處，做了上訴的處理。
在有多個叢集的情況下，而且兩個叢集都是正常的，那麼到底需要執行哪個？
/**
* 呼叫服務提供者
* @param invocation
* @return
* @throws RpcException
*/
@Override
public Result invoke(final Invocation invocation) throws RpcException {
// 校驗是否銷燬
checkWhetherDestroyed();
//TODO
// binding attachments into invocation.
Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
if (contextAttachments != null && contextAttachments.size() != 0) {
((RpcInvocation) invocation).addAttachments(contextAttachments);
}
// 獲得所有服務提供者 Invoker 集合
List<Invoker<T>> invokers = list(invocation);
// 獲得 LoadBalance 物件
LoadBalance loadbalance = initLoadBalance(invokers, invocation);
// 設定呼叫編號，若是非同步呼叫
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
// 執行呼叫
return doInvoke(invocation, invokers, loadbalance);
}/**
* 實現子 Cluster 的 Invoker 實現類的服務呼叫的差異邏輯，
* @param invocation
* @param invokers
* @param loadbalance
* @return
* @throws RpcException
*/
// 抽象方法，子類自行的實現 因為我們使用的預設配置，所以 我們將會是FailoverClusterInvoker 這個類
//< 8 doInvoker 方法>
protected abstract Result doInvoke(Invocation invocation, List<Invoker<T>> invokers,
LoadBalance loadbalance) throws RpcException;9 進入到 相應的叢集容錯方案 類中 因為我們使用的預設配置，所以 我們將會是FailoverClusterInvoker 這個類
/**
* 實際邏輯很簡單：迴圈，查詢一個 Invoker 物件，進行呼叫，直到成功
* @param invocation
* @param invokers
* @param loadbalance
* @return
* @throws RpcException
*/
@Override
@SuppressWarnings({"unchecked", "rawtypes"})
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
List<Invoker<T>> copyinvokers = invokers;
// 檢查copyinvokers即可用Invoker集合是否為空，如果為空，那麼丟擲異常
checkInvokers(copyinvokers, invocation);
String methodName = RpcUtils.getMethodName(invocation);
// 得到最大可呼叫次數：最大可重試次數+1，預設最大可重試次數Constants.DEFAULT_RETRIES=2
int len = getUrl().getMethodParameter(methodName, 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);
// failover機制核心實現：如果出現呼叫失敗，那麼重試其他伺服器
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列表已發生變化.
// 注意：如果列表發生了變化，那麼invoked判斷會失效，因為invoker示例已經改變
if (i > 0) {
checkWhetherDestroyed();
// 根據Invocation呼叫資訊從Directory中獲取所有可用Invoker
copyinvokers = list(invocation);
// check again
// 重新檢查一下
checkInvokers(copyinvokers, invocation);
}
// 根據負載均衡機制從copyinvokers中選擇一個Invoker
//< 9 select ------------------------- 下面將進行 invoker選擇----------------------------------->
Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
// 儲存每次呼叫的Invoker
invoked.add(invoker);
// 設定已經呼叫的 Invoker 集合，到 Context 中
RpcContext.getContext().setInvokers((List) invoked);
try {
// RPC 呼叫得到 Result
Result result = invoker.invoke(invocation);
// 重試過程中，將最後一次呼叫的異常資訊以 warn 級別日誌輸出
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;
}
// 其他性質的異常統一封裝成RpcException
le = e;
} catch (Throwable e) {
le = new RpcException(e.getMessage(), e);
} finally {
providers.add(invoker.getUrl().getAddress());
}
}
// 最大可呼叫次數用完還得到Result的話，丟擲RpcException異常：重試了N次還是失敗，並輸出最後一次異常資訊
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);
}9 select 使用 loadbalance 選擇 invoker
/**
* Select a invoker using loadbalance policy.</br>
* a) Firstly, select an invoker using loadbalance. If this invoker is in previously selected list, or,
* if this invoker is unavailable, then continue step b (reselect), otherwise return the first selected invoker</br>
* <p>
* b) Reselection, the validation rule for reselection: selected > available. This rule guarantees that
* the selected invoker has the minimum chance to be one in the previously selected list, and also
* guarantees this invoker is available.
*
* @param loadbalance load balance policy
* @param invocation invocation
* @param invokers invoker candidates
* @param selected exclude selected invokers or not
* @return the invoker which will final to do invoke.
* @throws RpcException
*/
/**
* 使用 loadbalance 選擇 invoker.
* a)
* @param loadbalance 物件，提供負責均衡策略
* @param invocation 物件
* @param invokers 候選的 Invoker 集合
* @param selected 已選過的 Invoker 集合. 注意：輸入保證不重複
* @return 最終的 Invoker 物件
* @throws RpcException
*/
protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.isEmpty())
return null;
// 獲得 sticky 配置項，方法級
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
// 若 stickyInvoker 不存在於 invokers 中，說明不在候選中，需要置空，重新選擇
if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
stickyInvoker = null;
}
//ignore concurrency problem
// 若開啟粘滯連線的特性，且 stickyInvoker 不存在於 selected 中，則返回 stickyInvoker 這個 Invoker 物件
if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
// 若開啟排除非可用的 Invoker 的特性，則校驗 stickyInvoker 是否可用。若可用，則進行返回
if (availablecheck && stickyInvoker.isAvailable()) {
return stickyInvoker;
}
}
}
// 執行選擇
//< 10 -----------------------------進行選擇------------------------------------->
Invoker<T> invoker = doSelect(loadbalance, invocation, invokers, selected);
// 若開啟粘滯連線的特性，記錄最終選擇的 Invoker 到 stickyInvoker
if (sticky) {
stickyInvoker = invoker;
}
return invoker;
}10 doSelect 從候選的 Invoker 集合，選擇一個最終呼叫的 Invoker 物件
/**
* 從候選的 Invoker 集合，選擇一個最終呼叫的 Invoker 物件
* @param loadbalance
* @param invocation
* @param invokers
* @param selected
* @return
* @throws RpcException
*/
private Invoker<T> doSelect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.isEmpty())
return null;
// 如果只有一個 Invoker ，直接選擇
if (invokers.size() == 1)
return invokers.get(0);
// 使用 Loadbalance ，選擇一個 Invoker 物件。
//<11 ------------------ 根據LoadBalance(負載均衡) 選擇一個合適的invoke>
Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);
//If the `invoker` is in the `selected` or invoker is unavailable && availablecheck is true, reselect.
// 如果 selected中包含（優先判斷） 或者 不可用&&availablecheck=true 則重試.
if ((selected != null && selected.contains(invoker))
|| (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
try {
//重選一個 Invoker 物件
Invoker<T> rinvoker = reselect(loadbalance, invocation, invokers, 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 = 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 的方法
* 各個負載均衡的類自行實現提供自定義的負載均衡策略。
* @param invokers
* @param url
* @param invocation
* @param <T>
* @return
*/
protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation);@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
int length = invokers.size(); // Number of invokers
int maxWeight = 0; // The maximum weight
int minWeight = Integer.MAX_VALUE; // The minimum weight
final LinkedHashMap<Invoker<T>, IntegerWrapper> invokerToWeightMap = new LinkedHashMap<Invoker<T>, IntegerWrapper>();
int weightSum = 0;
// 計算最小、最大權重，總的權重和。
for (int i = 0; i < length; i++) {
int weight = getWeight(invokers.get(i), invocation);
maxWeight = Math.max(maxWeight, weight); // Choose the maximum weight
minWeight = Math.min(minWeight, weight); // Choose the minimum weight
if (weight > 0) {
invokerToWeightMap.put(invokers.get(i), new IntegerWrapper(weight));
weightSum += weight;
}
}
// 計算最小、最大權重，總的權重和。
AtomicPositiveInteger sequence = sequences.get(key);
if (sequence == null) {
sequences.putIfAbsent(key, new AtomicPositiveInteger());
sequence = sequences.get(key);
}
// 獲得當前順序號，並遞增 + 1
int currentSequence = sequence.getAndIncrement();
// 權重不相等，順序根據權重分配
if (maxWeight > 0 && minWeight < maxWeight) {
int mod = currentSequence % weightSum;// 剩餘權重
for (int i = 0; i < maxWeight; i++) {// 迴圈最大權重
for (Map.Entry<Invoker<T>, IntegerWrapper> each : invokerToWeightMap.entrySet()) {
final Invoker<T> k = each.getKey();
final IntegerWrapper v = each.getValue();
// 剩餘權重歸 0 ，當前 Invoker 還有剩餘權重，返回該 Invoker 物件
if (mod == 0 && v.getValue() > 0) {
return k;
}
// 若 Invoker 還有權重值，扣除它( value )和剩餘權重( mod )。
if (v.getValue() > 0) {
v.decrement();
mod--;
}
}
}
}
// 權重相等，平均順序獲得
// Round robin
//<13 --------------------------->
return invokers.get(currentSequence % length);
}/**
* 實際邏輯很簡單：迴圈，查詢一個 Invoker 物件，進行呼叫，直到成功
* @param invocation
* @param invokers
* @param loadbalance
* @return
* @throws RpcException
*/
@Override
@SuppressWarnings({"unchecked", "rawtypes"})
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
List<Invoker<T>> copyinvokers = invokers;
// 檢查copyinvokers即可用Invoker集合是否為空，如果為空，那麼丟擲異常
checkInvokers(copyinvokers, invocation);
String methodName = RpcUtils.getMethodName(invocation);
// 得到最大可呼叫次數：最大可重試次數+1，預設最大可重試次數Constants.DEFAULT_RETRIES=2
int len = getUrl().getMethodParameter(methodName, 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);
// failover機制核心實現：如果出現呼叫失敗，那麼重試其他伺服器
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列表已發生變化.
// 注意：如果列表發生了變化，那麼invoked判斷會失效，因為invoker示例已經改變
if (i > 0) {
checkWhetherDestroyed();
// 根據Invocation呼叫資訊從Directory中獲取所有可用Invoker
copyinvokers = list(invocation);
// check again
// 重新檢查一下
checkInvokers(copyinvokers, invocation);
}
// 根據負載均衡機制從copyinvokers中選擇一個Invoker
//< ------------------------- 下面將進行 invoker選擇----------------------------------->
Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
// 儲存每次呼叫的Invoker
invoked.add(invoker);
// 設定已經呼叫的 Invoker 集合，到 Context 中
RpcContext.getContext().setInvokers((List) invoked);
try {
// RPC 呼叫得到 Result
Result result = invoker.invoke(invocation);
// 重試過程中，將最後一次呼叫的異常資訊以 warn 級別日誌輸出
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;
}
// 其他性質的異常統一封裝成RpcException
le = e;
} catch (Throwable e) {
le = new RpcException(e.getMessage(), e);
} finally {
providers.add(invoker.getUrl().getAddress());
}
}
// 最大可呼叫次數用完還得到Result的話，丟擲RpcException異常：重試了N次還是失敗，並輸出最後一次異常資訊
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);
}以上是一個完成呼叫過程的原始碼分析 以及架構分析
Dubbo叢集容錯原始碼解析
** 我們 後面的分析重點就是Cluster的呼叫過程**
Cluster的作用
Cluster 將 Directory 中的多個 Invoker 偽裝成一個 Invoker，對上層透明，偽裝過程包含了容錯邏輯，呼叫失敗後，重試另一個
應對出錯情況採取的策略,在某次出現錯誤後將會採用何種方式進行下次重試
叢集模式 概括
下面對這些一個一個分析
失敗自動切換，當出現失敗，重試其它伺服器 [1]。通常用於讀操作，但重試會帶來更長延遲。可通過 retries="2" 來設定重試次數(不含第一次)。
重試次數配置如下：
<dubbo:service retries="2" />或
<dubbo:reference retries="2" />或
<dubbo:reference>
<dubbo:method name="findFoo" retries="2" />
</dubbo:reference>核心呼叫類
/**
* 實際邏輯很簡單：迴圈，查詢一個 Invoker 物件，進行呼叫，直到成功
* @param invocation
* @param invokers
* @param loadbalance
* @return
* @throws RpcException
*/
@Override
@SuppressWarnings({"unchecked", "rawtypes"})
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
List<Invoker<T>> copyinvokers = invokers;
// 檢查copyinvokers即可用Invoker集合是否為空，如果為空，那麼丟擲異常
checkInvokers(copyinvokers, invocation);
String methodName = RpcUtils.getMethodName(invocation);
// 得到最大可呼叫次數：最大可重試次數+1，預設最大可重試次數Constants.DEFAULT_RETRIES=2
int len = getUrl().getMethodParameter(methodName, 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);
// failover機制核心實現：如果出現呼叫失敗，那麼重試其他伺服器
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列表已發生變化.
// 注意：如果列表發生了變化，那麼invoked判斷會失效，因為invoker示例已經改變
if (i > 0) {
checkWhetherDestroyed();
// 根據Invocation呼叫資訊從Directory中獲取所有可用Invoker
copyinvokers = list(invocation);
// check again
// 重新檢查一下
checkInvokers(copyinvokers, invocation);
}
// 根據負載均衡機制從copyinvokers中選擇一個Invoker
//< ------------------------- 下面將進行 invoker選擇----------------------------------->
Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
// 儲存每次呼叫的Invoker
invoked.add(invoker);
// 設定已經呼叫的 Invoker 集合，到 Context 中
RpcContext.getContext().setInvokers((List) invoked);
try {
// RPC 呼叫得到 Result
Result result = invoker.invoke(invocation);
// 重試過程中，將最後一次呼叫的異常資訊以 warn 級別日誌輸出
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;
}
// 其他性質的異常統一封裝成RpcException
le = e;
} catch (Throwable e) {
le = new RpcException(e.getMessage(), e);
} finally {
providers.add(invoker.getUrl().getAddress());
}
}
// 最大可呼叫次數用完還得到Result的話，丟擲RpcException異常：重試了N次還是失敗，並輸出最後一次異常資訊
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);
}快速失敗，只發起一次呼叫，失敗立即報錯。通常用於非冪等性的寫操作，比如新增記錄。
public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
// 檢查 invokers集合
checkInvokers(invokers, invocation);
// 根據負載均衡機制從 invokers 中選擇一個Invoker
Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
try {
// RPC 呼叫得到 Result
return invoker.invoke(invocation);
} catch (Throwable e) {
// 若是業務性質的異常，直接丟擲
if (e instanceof RpcException && ((RpcException) e).isBiz()) { // biz exception.
throw (RpcException) e;
}
// 封裝 RpcException 異常，並丟擲
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);
}
}失敗安全，出現異常時，直接忽略。通常用於寫入審計日誌等操作。
public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
try {
// 檢查 invokers 是否為空
checkInvokers(invokers, invocation);
// 根據負載均衡機制從 invokers 中選擇一個Invoker
Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
// RPC 呼叫得到 Result
return invoker.invoke(invocation);
} catch (Throwable e) {
// 列印異常日誌
logger.error("Failsafe ignore exception: " + e.getMessage(), e);
// 忽略異常
return new RpcResult(); // ignore
}
}失敗自動恢復，後臺記錄失敗請求，定時重發。通常用於訊息通知操作。
/**
* 當失敗的時候會將invocation 新增到失敗集合中
* @param invocation 失敗的invocation
* @param router 物件自身
*/
private void addFailed(Invocation invocation, AbstractClusterInvoker<?> router) {
//如果定時任務未初始化，進行建立
if (retryFuture == null) {
synchronized (this) {
if (retryFuture == null) {
retryFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
// collect retry statistics
//建立的定時任務，會呼叫 #retryFailed() 方法，重試任務，發起 RCP 呼叫。
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);
}
/**
* 重試任務，發起 RCP 呼叫
*/
void retryFailed() {
if (failed.size() == 0) {
return;
}
// 迴圈重試任務，逐個呼叫
for (Map.Entry<Invocation, AbstractClusterInvoker<?>> entry : new HashMap<>(failed).entrySet()) {
Invocation invocation = entry.getKey();
Invoker<?> invoker = entry.getValue();
try {
// RPC 呼叫得到 Result
invoker.invoke(invocation);
// 移除失敗任務
failed.remove(invocation);
} catch (Throwable e) {
logger.error("Failed retry to invoke method " + invocation.getMethodName() + ", waiting again.", e);
}
}
}
@Override
protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
try {
checkInvokers(invokers, invocation);
// 根據負載均衡機制從 invokers 中選擇一個Invoker
Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
// RPC 呼叫得到 Result
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
}
}並行呼叫多個伺服器，只要一個成功即返回。通常用於實時性要求較高的讀操作，但需要浪費更多服務資源。可通過 forks="2" 來設定最大並行數。
public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
try {
// 檢查 invokers 是否為空
checkInvokers(invokers, invocation);
// 儲存選擇的 Invoker 集合
final List<Invoker<T>> selected;
// 得到最大並行數，預設為 Constants.DEFAULT_FORKS = 2
final int forks = getUrl().getParameter(Constants.FORKS_KEY, Constants.DEFAULT_FORKS);
// 獲得呼叫超時時間，預設為 DEFAULT_TIMEOUT = 1000 毫秒
final int timeout = getUrl().getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
//如果最大並行數小於 0 或者大於invokers的數量，直接呼叫invokers
if (forks <= 0 || forks >= invokers.size()) {
selected = invokers;
} else {
// 迴圈，根據負載均衡機制從 invokers，中選擇一個個Invoker ，從而組成 Invoker 集合。
// 注意，因為增加了排重邏輯，所以不能保證獲得的 Invoker 集合的大小，小於最大並行數
selected = new ArrayList<>();
for (int i = 0; i < forks; i++) {
// TODO. Add some comment here, refer chinese version for more details.
// 在invoker列表(排除selected)後,如果沒有選夠,則存在重複迴圈問題.見select實現.
Invoker<T> invoker = select(loadbalance, invocation, invokers, selected);
if (!selected.contains(invoker)) {
//Avoid add the same invoker several times.
selected.add(invoker);
}
}
}
// 設定已經呼叫的 Invoker 集合，到 Context 中
RpcContext.getContext().setInvokers((List) selected);
// 異常計數器
final AtomicInteger count = new AtomicInteger();
// 建立阻塞佇列
final BlockingQueue<Object> ref = new LinkedBlockingQueue<>();
// 迴圈 selected 集合，提交執行緒池，發起 RPC 呼叫
for (final Invoker<T> invoker : selected) {
executor.execute(new Runnable() {
@Override
public void run() {
try {
// RPC 呼叫，獲得 Result 結果
Result result = invoker.invoke(invocation);
// 新增 Result 到 `ref` 阻塞佇列
ref.offer(result);
} catch (Throwable e) {
// 異常計數器 + 1
int value = count.incrementAndGet();
// 若 RPC 呼叫結果都是異常，則新增異常到 `ref` 阻塞佇列
if (value >= selected.size()) {
ref.offer(e);
}
}
}
});
}
try {
// 從 `ref` 佇列中，阻塞等待結果
Object ret = ref.poll(timeout, TimeUnit.MILLISECONDS);
// 若是異常結果，丟擲 RpcException 異常
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();
}
}廣播呼叫所有提供者，逐個呼叫，任意一臺報錯則報錯 [2]。通常用於通知所有提供者更新快取或日誌等本地資源資訊
public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
// 檢查 invokers 即可用Invoker集合是否為空，如果為空，那麼丟擲異常
checkInvokers(invokers, invocation);
// 設定已經呼叫的 Invoker 集合，到 Context 中
RpcContext.getContext().setInvokers((List) invokers);
// 儲存最後一次呼叫的異常
RpcException exception = null;
// 儲存最後一次呼叫的結果
Result result = null;
// 迴圈候選的 Invoker 集合，呼叫所有 Invoker 物件。
for (Invoker<T> invoker : invokers) {
try {
// 發起 RPC 呼叫
result = invoker.invoke(invocation);
} catch (RpcException e) {
exception = e;
logger.warn(e.getMessage(), e);
} catch (Throwable e) {
// 封裝成 RpcException 異常
exception = new RpcException(e.getMessage(), e);
logger.warn(e.getMessage(), e);
}
}
// 若存在一個異常，丟擲該異常
if (exception != null) {
throw exception;
}
return result;
}遍歷所有從Directory中list出來的Invoker集合，呼叫第一個isAvailable()的Invoker，只發起一次呼叫，失敗立即報錯。
isAvailable()判斷邏輯如下--Client處理連線狀態，且不是READONLY：
@Override
public boolean isAvailable() {
if (!super.isAvailable())
return false;
for (ExchangeClient client : clients){
if (client.isConnected() && !client.hasAttribute(Constants.CHANNEL_ATTRIBUTE_READONLY_KEY)){
//cannot write == not Available ?
return true ;
}
}
return false;
}