Dubbo源碼：叢集容錯
-
- 大緻流程
-
- FailoverClusterInvoker：失敗切換
- FailbackClusterInvoker：失敗恢複
- FailfastClusterInvoker：快速失敗
- FailsafeClusterInvoker：失敗安全
- ForkingClusterInvoker：并行調用多個服務提供者
- BroadcastClusterInvoker：會逐個調用每個服務提供者
大緻流程
消費者在生成代理對象後，調用RPC服務的方法，會執行代理對象的方法，因為建立代理對象的時候會傳入一個
InvocationHandler
對象，這個對象是代理對象的一個屬性，并且這個對象持有Invoker對象，
Invoker
對象是Dubbo領域模型中的核心模型，是實體域，包含這個RPC服務的所有資訊。
代理對象對象中其實就是調用這個Invoker的invoker()方法，是以就會執行到
AbstractClusterInvoker
類的invoker()方法，這個方法是叢集容錯的父類，裡面封裝了叢集容錯會用到的一些公共方法，比如調用list()從RegistryDirectory中擷取到擷取Invoker清單，選擇一個負載均衡器（LoadBalance），還有select()也就是負載均衡方法等等。
AbstractClusterInvoker還有一個doInvoke()
的模闆方法，具體實作調用邏輯和容錯邏輯子類實作。下面就來看下它的幾個主要實作類。
FailoverClusterInvoker：失敗切換
是預設的容錯機制，會先根據配置擷取配置的重試次數，預設是3次。然後循環調用，如果不是第一次調用會調用list()重選列舉服務目錄。然後負載均衡選擇Invoker，并且把選中後的Invoker記錄下來，當失敗重試的時候如果再次選到，會重新選擇一次。最後調用Invoker的invoker()，成功則傳回結果。如果調用失敗會記錄異常資訊，然後重試。直到超出重試次數後，會抛出異常，告訴服務調用失敗了。
通常用于讀操作，但重試會帶來更長延遲。 可通過retries="2"來設定重試次數（不含第一次）。
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
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++) {
if (i > 0) {
checkWhetherDestroyed();
// 在進行重試前重新列舉 Invoker，這樣做的好處是，如果某個服務挂了，
// 通過調用 list 可得到最新可用的 Invoker 清單
copyinvokers = list(invocation);
// check again
checkInvokers(copyinvokers, invocation);
}
// 通過負載均衡選擇 Invoker
Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
// 選擇後的記錄下來，失敗重試如果選到的在這裡面會進行重選
invoked.add(invoker);
RpcContext.getContext().setInvokers((List) invoked);
try {
// 調用
Result result = invoker.invoke(invocation);
if (le != null && logger.isWarnEnabled()) {
logger.warn("....");
}
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("....");
}
FailbackClusterInvoker：失敗恢複
會在調用失敗後，把任務放入到一個ConcurrentMap中，并啟動一個延時任務去消費這個Map，最後傳回一個空結果給服務消費者。适合執行消息通知等操作。
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);
// 調用
return invoker.invoke(invocation);
} catch (Throwable e) {
// 如果調用過程中發生異常，此時僅列印錯誤日志，不抛出異常
logger.error("...");
// 添加到失敗清單中，定時重試
addFailed(invocation, this);
// 傳回空結果
return new RpcResult(); // ignore
}
}
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("....");
}
}
}
FailfastClusterInvoker：快速失敗
隻會進行一次調用，失敗後立即抛出異常。适用于幂等操作，比如新增記錄。
public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
checkInvokers(invokers, invocation);
Invoker<T> invoker = select(loadbalance, invocation, invokers, 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);
}
}
FailsafeClusterInvoker：失敗安全
當調用過程中出現異常時，FailsafeClusterInvoker 僅會列印異常，而不會抛出異常。适用于寫入審計日志等操作。
public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
try {
checkInvokers(invokers, invocation);
Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
return invoker.invoke(invocation);
} catch (Throwable e) {
logger.error("Failsafe ignore exception: " + e.getMessage(), e);
return new RpcResult(); // ignore
}
}
ForkingClusterInvoker：并行調用多個服務提供者
會在運作時通過線程池建立多個線程，并發調用多個服務提供者，然後把結果放入BlockingQueue中，如果所有都調用失敗，Queue中放的就是失敗的異常資訊。隻要有一個服務提供者成功傳回了，Queue中放的就是結果，最後從阻塞隊列中取結果，如果取到的結果是異常就抛出，是結果就正常傳回。
主要應用場景是在一些對實時性要求比較高讀操作。
public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
checkInvokers(invokers, invocation);
final List<Invoker<T>> selected;
final int forks = getUrl().getParameter(Constants.FORKS_KEY, Constants.DEFAULT_FORKS);
final int timeout = getUrl().getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
if (forks <= 0 || forks >= invokers.size()) {
selected = invokers;
} else {
selected = new ArrayList<Invoker<T>>();
for (int i = 0; i < forks; i++) {
// TODO. Add some comment here, refer chinese version for more details.
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);
// 記錄失敗次數
final AtomicInteger count = new AtomicInteger();
// 記錄成功或者異常資訊
final BlockingQueue<Object> ref = new LinkedBlockingQueue<Object>();
for (final Invoker<T> invoker : selected) {
executor.execute(new Runnable() {
public void run() {
try {
// 執行
Result result = invoker.invoke(invocation);
ref.offer(result);
} catch (Throwable e) {
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);
}
return (Result) ret;
} catch (InterruptedException e) {
throw new RpcException("");
}
}
BroadcastClusterInvoker：會逐個調用每個服務提供者
如果其中一台報錯，在循環調用結束後，BroadcastClusterInvoker 會抛出異常。該類通常用于通知所有提供者更新緩存或日志等本地資源資訊
public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
checkInvokers(invokers, invocation);
RpcContext.getContext().setInvokers((List) invokers);
RpcException exception = null;
Result result = null;
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;
}
![初識dubbo整合spring出現的service注入失敗的問題[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)