前言
image
上一篇降到movk本地伪装其实是对Cluster集群容错类实例的再次包装,调完mockInvoker的invoker方法后,会调到被包装的Cluster实例的invoker,对真正的远程调用执行各种集群容错的策略。
image
集群容错策略的选择
- 配置方式 ：指定集群容错策略为available`
@Reference(cluster="AvailableCluster")
private UserService userService;
- 默认策略 ：failover
Cluster实例都是从SPI工厂里加载而来的,如果没有主动配置,默认得到的就是接口上@SPI的值
image
Cluster$Adaptive源代码 ：
public class Cluster$Adaptive implements com.alibaba.dubbo.rpc.cluster.Cluster {
public com.alibaba.dubbo.rpc.Invoker join(com.alibaba.dubbo.rpc.cluster.Directory arg0) throws com.alibaba.dubbo.rpc.RpcException {
if (arg0 == null)
throw new IllegalArgumentException("com.alibaba.dubbo.rpc.cluster.Directory argument == null");
if (arg0.getUrl() == null)
throw new IllegalArgumentException("com.alibaba.dubbo.rpc.cluster.Directory argument getUrl() == null");
// 从url拿cluster的配置,没有配就failover
com.alibaba.dubbo.common.URL url = arg0.getUrl();
String extName = url.getParameter("cluster", "failover");
if (extName == null)
throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.cluster.Cluster) name from url(" + url.toString() + ") use keys([cluster])");
com.alibaba.dubbo.rpc.cluster.Cluster extension = (com.alibaba.dubbo.rpc.cluster.Cluster) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.cluster.Cluster.class).getExtension(extName);
return extension.join(arg0);
}
}
FailoverClusterInvoker
当首次调用失败后,还会重试指定次数
循环最大调用次数（ 1+重试次数）,每次都去发起远程调用,如果其中一次调用成功就return,结束循环,否则继续下一次调用如果超出重试次数还没有调用成功,就抛rpc异常
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,Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
if (len <= 0) {
len = 1;
}
// retry loop.
RpcException le = null; // last exception.
//已经调用过了的服务列表
List<Invoker<T>> invoked = newArrayList<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) {
checkWhetherDestroyed();
copyinvokers = list(invocation);
// check again
checkInvokers(copyinvokers, invocation);
}
//根据负载均衡算法，选择一个服务调用
Invoker<T> invoker = select(loadbalance, invocation, copyinvokers,invoked);
//记录已经调用过的invoker
invoked.add(invoker);
RpcContext.getContext().setInvokers((List) invoked);
try {
//具体的服务调用逻辑
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());
}
}
// 如果超出重试次数还没有调用成功,就抛rpc异常
throw new RpcException(le != null ? le.getCode() : 0, "Failed toinvoke 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 thedubbo version "
+ Version.getVersion() + ". Last error is: "
+ (le != null ? le.getMessage() : ""), le != null && le.getCause() != null ? le.getCause() : le);
}
FailBackClusterInvoker
出错返回空结果,异步重试，起一个线程，等待配置的时间,进行重试
catch到异常
image
image
FailfastClusterInvoker
catch到异常并直接抛出RpcException，不进行重试,以快速响应
image
FailsafeClusterInvoker
直接吞掉异常,不进行任何处理,返回空的结果,适用于对结果不敏感的业务
image
ForkingClusterInvoker
同时调n台(默认为2台)主机，选择最快返回结果的那台主机返回的结果
负载均衡后选出与fork数相同的invoker
image
selecte出几个invoker就起对应数量的县城,去发起远程调用,并异步把结果放到LinkedBlockingQueue 阻塞队列中
image
一旦从队列poll到第一个result,就return 结果
image
BroadcastClusterInvoker
服务列表里所有主机都会被调,原子广播
image
AvailableClusterInvoker
没有负载均衡没有重试，调服务之前会先查看tcp存在有效的长连接,确保后端服务正常,如果有,再调用具体的业务方法。用于对数据安全要求比较高的业务
image
image
