前兩篇中,我們看到了dubbo在負載均衡和服務路由方面的實作,它為叢集功能提供了必要的功能。
今天我們再來看另一個叢集元件的實作:叢集容錯。
1. dubbo 叢集容錯簡介
為了避免單點故障,現在的應用通常至少會部署在兩台伺服器上。對于一些負載比較高的服務,會部署更多的伺服器。對于服務消費者來說,同一環境下出現了多個服務提供者。這時會出現一個問題,服務消費者需要決定選擇哪個服務提供者進行調用。另外服務調用失敗時的處理措施也是需要考慮的,是重試呢,還是抛出異常,亦或是隻列印異常等。為了處理這些問題,Dubbo 定義了叢集接口 Cluster 以及 Cluster Invoker。叢集 Cluster 用途是将多個服務提供者合并為一個 Cluster Invoker,并将這個 Invoker 暴露給服務消費者。這樣一來,服務消費者隻需通過這個 Invoker 進行遠端調用即可,至于具體調用哪個服務提供者,以及調用失敗後如何處理等問題,現在都交給叢集子產品去處理。叢集子產品是服務提供者和服務消費者的中間層,為服務消費者屏蔽了服務提供者的情況,這樣服務消費者就可以專心處理遠端調用相關事宜。
dubbo的叢集容錯功能由多個元件共同完成:包括 Cluster、Cluster Invoker、Directory、Router 和 LoadBalance 等。它們之間的依賴關系如下:
負載均衡、路由服務是在一次調用中進行的,而容錯則是當調用發生異常之後,進行處理政策。
dubbo中主要提供了以下幾種容錯政策實作:
Failover Cluster - 失敗自動切換
Failfast Cluster - 快速失敗
Failsafe Cluster - 失敗安全
Failback Cluster - 失敗自動恢複
Forking Cluster - 并行調用多個服務提供者
以上叢集容錯政策可以通過提供者或者消費者的 service或reference進行配置:
<dubbo:service cluster="failsafe" />
<dubbo:reference cluster="failsafe" />
其優先級同樣遵循dubbo設計原則,消費端配置優先,其次是提供端。不配置情況下預設是failover政策,預設重試3次。
2. 叢集容錯的架構實作
叢集接口 Cluster 和 Cluster Invoker,這兩者是不同的。Cluster 是接口,而 Cluster Invoker 是一種 Invoker。服務提供者的選擇邏輯,以及遠端調用失敗後的的處理邏輯均是封裝在 Cluster Invoker 中。
Cluster 的實作類圖如下:
各個Cluster的實作都很簡單,也都統一繼承了 AbstractCluster, 而該 AbstractCluster 則做了一層統一的攔截器的功能接入,實作如下:
public abstract class AbstractCluster implements Cluster {
private <T> Invoker<T> buildClusterInterceptors(AbstractClusterInvoker<T> clusterInvoker, String key) {
AbstractClusterInvoker<T> last = clusterInvoker;
List<ClusterInterceptor> interceptors = ExtensionLoader.getExtensionLoader(ClusterInterceptor.class).getActivateExtension(clusterInvoker.getUrl(), key);
// 根據需要包裝ClusterInvoker, 使用切面的方式進行攔截器接入
// 按先後依次強入攔截器
if (!interceptors.isEmpty()) {
for (int i = interceptors.size() - 1; i >= 0; i--) {
final ClusterInterceptor interceptor = interceptors.get(i);
final AbstractClusterInvoker<T> next = last;
// 使用内部類進行包裝攔截器
// 先後順序如: beforeC -> beforeB -> beforeA (spring中還有Around) -> afterA -> afterB -> afterC (spring中還有afterReturn)
last = new InterceptorInvokerNode<>(clusterInvoker, interceptor, next);
}
}
return last;
}
@Override
public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
// ClusterInvoker 調用入口, 讓具體政策實作 doJoin(), 并在其基礎上進行包裝攔截器, 依據來源 reference.interceptor=xxx
return buildClusterInterceptors(doJoin(directory), directory.getUrl().getParameter(REFERENCE_INTERCEPTOR_KEY));
}
//
protected abstract <T> AbstractClusterInvoker<T> doJoin(Directory<T> directory) throws RpcException;
protected class InterceptorInvokerNode<T> extends AbstractClusterInvoker<T> {
private AbstractClusterInvoker<T> clusterInvoker;
private ClusterInterceptor interceptor;
private AbstractClusterInvoker<T> next;
public InterceptorInvokerNode(AbstractClusterInvoker<T> clusterInvoker,
ClusterInterceptor interceptor,
AbstractClusterInvoker<T> next) {
this.clusterInvoker = clusterInvoker;
this.interceptor = interceptor;
this.next = next;
}
@Override
public Class<T> getInterface() {
return clusterInvoker.getInterface();
}
@Override
public URL getUrl() {
return clusterInvoker.getUrl();
}
@Override
public boolean isAvailable() {
return clusterInvoker.isAvailable();
}
@Override
public Result invoke(Invocation invocation) throws RpcException {
Result asyncResult;
try {
// 攔截器的具體處理邏輯
// 有個 intercept() 的預設方法,其為調用 clusterInvoker.invoke(invocation); 進而實作鍊式調用
interceptor.before(next, invocation);
asyncResult = interceptor.intercept(next, invocation);
} catch (Exception e) {
// onError callback
if (interceptor instanceof ClusterInterceptor.Listener) {
ClusterInterceptor.Listener listener = (ClusterInterceptor.Listener) interceptor;
listener.onError(e, clusterInvoker, invocation);
}
throw e;
} finally {
interceptor.after(next, invocation);
}
return asyncResult.whenCompleteWithContext((r, t) -> {
// onResponse callback
if (interceptor instanceof ClusterInterceptor.Listener) {
ClusterInterceptor.Listener listener = (ClusterInterceptor.Listener) interceptor;
if (t == null) {
listener.onMessage(r, clusterInvoker, invocation);
} else {
listener.onError(t, clusterInvoker, invocation);
}
}
});
}
@Override
public void destroy() {
clusterInvoker.destroy();
}
@Override
public String toString() {
return clusterInvoker.toString();
}
@Override
protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
// The only purpose is to build a interceptor chain, so the cluster related logic doesn't matter.
return null;
}
}
}
接下來,我們詳細看看,每個叢集容錯政策都是如何建立的。
// failover 失敗自動切換
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);
}
}
// failfast 快速失敗
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);
}
}
// failsafe 失敗安全
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);
}
}
// failback 失敗自動恢複
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);
}
}
// forking 并行調用多個服務提供者
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);
}
}
// mergeable 合并結果容錯
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);
}
}
3. 具體叢集容錯的實作
failover, 失敗自動切換。這是dubbo的預設叢集容錯政策,因為它是一個比較通用的政策,即隻需做重試即可,保證高可用。
整個叢集容錯政策的調用入口在 AbstractClusterInvoker.invoke() 中,經過一些通用過程調用後,再由具體政策實作 doInvoke();
// org.apache.dubbo.rpc.cluster.support.AbstractClusterInvoker#invoke
@Override
public Result invoke(final Invocation invocation) throws RpcException {
// 有效性檢查
checkWhetherDestroyed();
// binding attachments into invocation.
Map<String, Object> contextAttachments = RpcContext.getContext().getObjectAttachments();
if (contextAttachments != null && contextAttachments.size() != 0) {
((RpcInvocation) invocation).addObjectAttachments(contextAttachments);
}
// 路由服務提供所有的 invokers
List<Invoker<T>> invokers = list(invocation);
// 擷取負載均衡器
LoadBalance loadbalance = initLoadBalance(invokers, invocation);
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
// 各子類實作 具體的容錯邏輯
return doInvoke(invocation, invokers, loadbalance);
}
各ClusterInvoker的實作類圖如下:
3.1. failover 失敗自動切換實作
// org.apache.dubbo.rpc.cluster.support.FailoverClusterInvoker#doInvoke
@Override
@SuppressWarnings({"unchecked", "rawtypes"})
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
List<Invoker<T>> copyInvokers = invokers;
checkInvokers(copyInvokers, invocation);
String methodName = RpcUtils.getMethodName(invocation);
int len = getUrl().getMethodParameter(methodName, RETRIES_KEY, 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.
if (i > 0) {
// 進行重試時,需要重新整理invokers
checkWhetherDestroyed();
copyInvokers = list(invocation);
// check again
checkInvokers(copyInvokers, invocation);
}
// 使用負載均衡選取一個 invoker
Invoker<T> invoker = select(loadbalance, invocation, copyInvokers, invoked);
// 将選中的invoker添加到 invoked 中,避免反複選擇一個失效的invoker
invoked.add(invoker);
RpcContext.getContext().setInvokers((List) invoked);
try {
// 調用選中的invoker 遠端服務,成功直接傳回了,失敗則容錯能力上
Result result = invoker.invoke(invocation);
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 {
providers.add(invoker.getUrl().getAddress());
}
}
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);
}
總結:failover 容錯,即是自動重試各可用提供者的過程。
3.2. failback 失敗自動恢複的實作
public FailbackClusterInvoker(Directory<T> directory) {
super(directory);
// retries=3
int retriesConfig = getUrl().getParameter(RETRIES_KEY, DEFAULT_FAILBACK_TIMES);
if (retriesConfig <= 0) {
retriesConfig = DEFAULT_FAILBACK_TIMES;
}
// failbacktasks=100
int failbackTasksConfig = getUrl().getParameter(FAIL_BACK_TASKS_KEY, DEFAULT_FAILBACK_TASKS);
if (failbackTasksConfig <= 0) {
failbackTasksConfig = DEFAULT_FAILBACK_TASKS;
}
retries = retriesConfig;
failbackTasks = failbackTasksConfig;
}
// 當調用失敗後,将其添加到定時隊列中,稍後進行重新請求
private void addFailed(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, Invoker<T> lastInvoker) {
if (failTimer == null) {
synchronized (this) {
if (failTimer == null) {
// 以1秒為間隔使用 hash環,掃描任務
failTimer = new HashedWheelTimer(
new NamedThreadFactory("failback-cluster-timer", true),
1,
TimeUnit.SECONDS, 32, failbackTasks);
}
}
}
// 使用 RetryTimerTask 來建構排程的任務
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());
}
}
@Override
protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
Invoker<T> invoker = null;
try {
checkInvokers(invokers, invocation);
invoker = select(loadbalance, invocation, invokers, 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(loadbalance, invocation, invokers, invoker);
return AsyncRpcResult.newDefaultAsyncResult(null, null, invocation); // ignore
}
}
總結:failback 容錯,即是隻做一次調用,失敗後會開啟後續定時任務進行重新調用的過程。
3.3. failfast 快速失敗的實作
// org.apache.dubbo.rpc.cluster.support.FailfastClusterInvoker#doInvoke
@Override
public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
checkInvokers(invokers, invocation);
// 使用負載均衡選取一個 可用的 invoker, 然後進行調用即可
// selected = null, 即隻一次選擇即可完成select
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 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);
}
}
// org.apache.dubbo.rpc.cluster.support.AbstractClusterInvoker#select
/**
* 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 exception
*/
protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation,
List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (CollectionUtils.isEmpty(invokers)) {
return null;
}
String methodName = invocation == null ? StringUtils.EMPTY_STRING : invocation.getMethodName();
boolean sticky = invokers.get(0).getUrl()
.getMethodParameter(methodName, CLUSTER_STICKY_KEY, DEFAULT_CLUSTER_STICKY);
//ignore overloaded method
if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
stickyInvoker = null;
}
//ignore concurrency problem
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;
}
private Invoker<T> doSelect(LoadBalance loadbalance, Invocation invocation,
List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (CollectionUtils.isEmpty(invokers)) {
return null;
}
if (invokers.size() == 1) {
return invokers.get(0);
}
Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);
//If the `invoker` is in the `selected` or invoker is unavailable && availablecheck is true, reselect.
if ((selected != null && selected.contains(invoker))
|| (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
try {
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.
int index = invokers.indexOf(invoker);
try {
//Avoid collision
invoker = invokers.get((index + 1) % invokers.size());
} 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;
}
總結: failfast 容錯,使用負載均衡政策選擇一次可用的invoker, 進行調用, 異常則抛出,正常則傳回結果。
3.4. failsafe 安全失敗容錯的實作
@Override
public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
try {
checkInvokers(invokers, invocation);
// 與failfast 一樣,隻使用一次負載均衡政策,選擇一個invoker調用即可
// 差别在于傳回值,failsafe 不抛出異常,當發生異常時傳回一個預設值
Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
return invoker.invoke(invocation);
} catch (Throwable e) {
logger.error("Failsafe ignore exception: " + e.getMessage(), e);
// 将異常資訊忽略,傳回預設值
return AsyncRpcResult.newDefaultAsyncResult(null, null, invocation); // ignore
}
}
總結: failsafe 容錯,即忽略掉所有異常,隻傳回正式結果。當發生異常時,傳回 AsyncRpcResult.newDefaultAsyncResult 作為結果,好像沒有發生異常一樣。
3.5. forking 并發請求容錯實作
// org.apache.dubbo.rpc.cluster.support.ForkingClusterInvoker#doInvoke
@Override
@SuppressWarnings({"unchecked", "rawtypes"})
public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
try {
checkInvokers(invokers, invocation);
final List<Invoker<T>> selected;
// forks=2
final int forks = getUrl().getParameter(FORKS_KEY, DEFAULT_FORKS);
// timeout=1000
final int timeout = getUrl().getParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT);
if (forks <= 0 || forks >= invokers.size()) {
selected = invokers;
} else {
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);
final AtomicInteger count = new AtomicInteger();
final BlockingQueue<Object> ref = new LinkedBlockingQueue<>();
for (final Invoker<T> invoker : selected) {
// 使用線程池進行并發調用 invoker
// 線程池為無界隊列式: executor = Executors.newCachedThreadPool(new NamedInternalThreadFactory("forking-cluster-timer", true));
executor.execute(() -> {
try {
Result result = invoker.invoke(invocation);
// 隻要結果響應,則入隊到 ref 中
ref.offer(result);
} catch (Throwable e) {
int value = count.incrementAndGet();
if (value >= selected.size()) {
// 當超過forks 數量的異常發生後,将異常資訊寫入ref中,即外部可以擷取結果了
ref.offer(e);
}
}
});
}
try {
// 阻塞擷取結果,最長等待 timeout
// 擷取第一個結果作為響應依據
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();
}
}
總結: forking 容錯,即是同時發起n個并發請求調用提供者,誰最先響應則傳回誰的結果。其他結果則全部忽略。可以說是非常耗資源的一種方式了,不過總是有相應的應用場景,是以存在。
3.6. broadcast 廣播容錯的實作
// org.apache.dubbo.rpc.cluster.support.BroadcastClusterInvoker#doInvoke
@Override
@SuppressWarnings({"unchecked", "rawtypes"})
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;
// 向所有invoker發起調用,隻要有一個異常,則抛出異常
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;
}
總結: broadcast 容錯,即向所有invoker發起調用(即廣播),全部成功才算成功。
3.7. mergeable 歸并容錯的實作
// org.apache.dubbo.rpc.cluster.support.MergeableClusterInvoker#doInvoke
@Override
protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
checkInvokers(invokers, invocation);
// merger=xxx
String merger = getUrl().getMethodParameter(invocation.getMethodName(), MERGER_KEY);
// 沒有指定merger, 直接調用一個可用 invoker 即可
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;
}
}
}
}
// 最後嘗試使用第一個 invoker.invoke()
return invokers.iterator().next().invoke(invocation);
}
Class<?> returnType;
try {
returnType = getInterface().getMethod(
invocation.getMethodName(), invocation.getParameterTypes()).getReturnType();
} catch (NoSuchMethodException e) {
returnType = null;
}
Map<String, Result> results = new HashMap<>();
for (final Invoker<T> invoker : invokers) {
RpcInvocation subInvocation = new RpcInvocation(invocation, invoker);
subInvocation.setAttachment(ASYNC_KEY, "true");
// 異步調用所有 invoker
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 {
// 等待所有 invoker 的結果響應
Result r = asyncResult.get();
if (r.hasException()) {
log.error("Invoke " + getGroupDescFromServiceKey(entry.getKey()) +
" failed: " + r.getException().getMessage(),
r.getException());
} else {
// 将所有結果放到 resultList 中
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) {
// 隻有一個結果,則傳回一個 Result
return resultList.iterator().next();
}
if (returnType == void.class) {
return AsyncRpcResult.newDefaultAsyncResult(invocation);
}
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);
}
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) {
method.invoke(result, r.getValue());
}
}
} catch (Exception e) {
throw new RpcException("Can not merge result: " + e.getMessage(), e);
}
} else {
Merger resultMerger;
// 解析出 merger, 調用 其 merge 方法,傳回結果
if (ConfigUtils.isDefault(merger)) {
resultMerger = MergerFactory.getMerger(returnType);
} else {
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());
}
// 有很多merger, 都在 org.apache.dubbo.rpc.cluster.merger中,
// 如: MapMerger/Array/Boolean/Int/List/Set/ByteArray...
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);
}
總結: mergeable 容錯,依次調用所有invokers, 并通過使用一個merger進行結果合并處理以傳回結果。雖然不知道有啥用,但是感覺很厲害的樣子。
dubbo的叢集容錯實作中,使用了 模闆方式模式,責任鍊模式,工廠模式,代理模式,使得各個容錯的實作顯得相當簡潔明了和簡單容易。這就是優秀架構的特性吧。
不要害怕今日的苦,你要相信明天,更苦!