1. dubbo集群容错简单介绍
参照官方文档的设计图：https://dubbo.gitbooks.io/dubbo-user-book/content/demos/fault-tolerent-strategy.html
官网设计图.png
第一次看这张图的时候，我没看明白，知道去看源码，然后回来再看这幅图，再去看源码就稍微清晰点了。
这里主要有五部分：
- Invoker
Invoker是服务调用者调用服务的一个抽象对象。里面封装了地址url以及接口信息。 - Cluster
是把所有Invoker封装成一个Invoker，以供服务调用者透明化调用，里面实际包含了容错逻辑等。 - Directory
Directory实际上是一个List<Invoker>，可以是静态的，也可以是动态的，比如注册中心推送变更时就是动态变化的。 - Router
Router根据路由规则从List<Invoker>中筛选满足路由规则的。 - LoadBalance
LoadBanlanc通过负载均衡策略选择出一个服务提供者（Invoker）来提供服务。
2. 源码分析dubbo集群容错
在分析源码之前，我先放一张我自己话的dubbo集群容错的时序图：
集群容错UML时序图.jpg
然后我们来根据这个时序图一步一步跟进源码。
2.1 InvokerInvocationHandler.class
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]);
}
return invoker.invoke(new RpcInvocation(method, args)).recreate();----------------------(1)invoke
}
现在我们要开始跟进了。
2.2开始第一步（1），进入到AbstractClusterInvoker.class
public Result invoke(final Invocation invocation) throws RpcException {
checkWheatherDestoried();
LoadBalance loadbalance;
List<Invoker<T>> invokers = list(invocation);--------------------------------------------（2）list invokers
if (invokers != null && invokers.size() > 0) {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
.getMethodParameter(invocation.getMethodName(),Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
} else {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
}
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
return doInvoke(invocation, invokers, loadbalance);
}
protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
List<Invoker<T>> invokers = directory.list(invocation);//从AbstractDirectory中获取
return invokers;
}
2.3 第二步(2)获取所有集群上的List<Invoker<T>>，AbstractDirectory.class
public List<Invoker<T>> list(Invocation invocation) throws RpcException {
if (destroyed){
throw new RpcException("Directory already destroyed .url: "+ getUrl());
}
List<Invoker<T>> invokers = doList(invocation);-----------(3)doList
List<Router> localRouters = this.routers; // local reference
if (localRouters != null && localRouters.size() > 0) {
for (Router router: localRouters){
try {
if (router.getUrl() == null || router.getUrl().getParameter(Constants.RUNTIME_KEY, true)) {
invokers = router.route(invokers, getConsumerUrl(), invocation);------------(5)route
}
} catch (Throwable t) {
logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
}
}
}
return invokers;
}
2.4 第三步（3）获取invoker，实际是通过RegistryDirectory.class或者StaticDirectory.class完成的。
public List<Invoker<T>> doList(Invocation invocation) {
if (forbidden) {
throw new RpcException(RpcException.FORBIDDEN_EXCEPTION, "Forbid consumer " + NetUtils.getLocalHost() + " access service " + getInterface().getName() + " from registry " + getUrl().getAddress() + " use dubbo version " + Version.getVersion() + ", Please check registry access list (whitelist/blacklist).");
}
List<Invoker<T>> invokers = null;
Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap; // local reference
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]); // 可根据第一个参数枚举路由
}
if(invokers == null) {
invokers = localMethodInvokerMap.get(methodName);
}
if(invokers == null) {
invokers = localMethodInvokerMap.get(Constants.ANY_VALUE);
}
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;
}
这里从localMethodInvokerMap中获取所有的invokers.
2.5第四步，已经获取到所有的Invoker了，就需要根据路由规则筛选出符合路由规则的Invoker。ConditionRouter.class
public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation)
throws RpcException {
if (invokers == null || invokers.size() == 0) {
return invokers;
}
try {
if (! matchWhen(url)) {
return invokers;
}
List<Invoker<T>> result = new ArrayList<Invoker<T>>();
if (thenCondition == null) {
logger.warn("The current consumer in the service blacklist. consumer: " + NetUtils.getLocalHost() + ", service: " + url.getServiceKey());
return result;
}
for (Invoker<T> invoker : invokers) {
if (matchThen(invoker.getUrl(), url)) {
result.add(invoker);
}
}
if (result.size() > 0) {
return result;
} else if (force) {
logger.warn("The route result is empty and force execute. consumer: " + NetUtils.getLocalHost() + ", service: " + url.getServiceKey() + ", router: " + url.getParameterAndDecoded(Constants.RULE_KEY));
return result;
}
} catch (Throwable t) {
logger.error("Failed to execute condition router rule: " + getUrl() + ", invokers: " + invokers + ", cause: " + t.getMessage(), t);
}
return invokers;
}
回到AbstractClusterInvoker.class，我们已经经历了两个过程：
- 获取所有集群上的invoker(list)
- 路由出所有正常能用的invoker(list)
2.5现在我们有了能够调用的List<Inovker>,到底选择哪一台服务器的Invoker给Consumer调用呢？dubbo官方规定在集群调用失败时，Dubbo 提供了多种容错方案，缺省为 failover 重试。我们来看看接下来有什么操作？
public Result invoke(final Invocation invocation) throws RpcException {
checkWheatherDestoried();
LoadBalance loadbalance;
List<Invoker<T>> invokers = list(invocation);
if (invokers != null && invokers.size() > 0) {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
.getMethodParameter(invocation.getMethodName(),Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
} else {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
}
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
return doInvoke(invocation, invokers, loadbalance); ----------------------------------doInvoke
}
现在我们已经有List<Inovker>了，选定某个服务提供者进行远程调用，但是服务调用可能出错，dubbo官方为我们提供了Failover（失败自动切换），Failfast（快速失败），Failsafe（失败安全），Failback（失败自动恢复），Forking（并行调用），Broadcast （广播调用），我们主要来看看默认的Failover容错机制，如何实现doInvoke.
2.6Failover容错机制，FailoverClusterInvoker.class
@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);
int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1; --------------------------@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++) { ------------------------------------------------@2
//重试时，进行重新选择，避免重试时invoker列表已发生变化.
//注意：如果列表发生了变化，那么invoked判断会失效，因为invoker示例已经改变
if (i > 0) {
checkWheatherDestoried();
copyinvokers = list(invocation); ----------------------------------------------@3
//重新检查一下
checkInvokers(copyinvokers, invocation);
}
Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked); --------------- @4
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());
}
}
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);
}
我们大概来分析一下这种失败重试的原理。
@1，获得重试的最大次数，默认为2次。实际这里加1，如果retries=2,这里len为3次，跟dubbo官方规定一样，是不包含第一次调用的。一定要记住，不需要失败重试的时候将其设置为0.
@2，这里开始循环执行重试，如果retries=0,则执行一次这里的for循环，将不再执行第二次。
@3，重试时，进行重新选择，避免重试时invoker列表已发生变化
@4，使用loadbalance选择invoker。
private Invoker<T> doselect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.size() == 0)
return null;
if (invokers.size() == 1)
return invokers.get(0);
// 如果只有两个invoker，退化成轮循
if (invokers.size() == 2 && selected != null && selected.size() > 0) {------------------------------------------@!
return selected.get(0) == invokers.get(0) ? invokers.get(1) : invokers.get(0);
}
Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);--------------------------------@2
//如果 selected中包含（优先判断） 或者 不可用&&availablecheck=true 则重试.
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{
//看下第一次选的位置，如果不是最后，选+1位置.
int index = invokers.indexOf(invoker);
try{
//最后在避免碰撞
invoker = index <invokers.size()-1?invokers.get(index+1) :invoker;
}catch (Exception e) {
logger.warn(e.getMessage()+" may because invokers list dynamic change, ignore.",e);
}
}
}catch (Throwable t){
logger.error("clustor relselect fail reason is :"+t.getMessage() +" if can not slove ,you can set cluster.availablecheck=false in url",t);
}
}
return invoker;
}
以上代码是通过负载均衡策略选择出一台可用的invoker,如果只有两个invoker，退化成轮循。默认的是随机策略。
2.6通过负载均衡选择一台可用的invoker。其中这里Random属于默认的负载均衡策略。负载均衡包括，Random（随机），RoundRobin(轮询)，ListActive(最少活跃数)，ConsistentHash（一致性hash），这里我们看默认的RandomLoadBalance.class
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
int length = invokers.size(); // 总个数
int totalWeight = 0; // 总权重
boolean sameWeight = true; // 权重是否都一样
for (int i = 0; i < length; i++) {
int weight = getWeight(invokers.get(i), invocation);
totalWeight += weight; // 累计总权重
if (sameWeight && i > 0
&& weight != getWeight(invokers.get(i - 1), invocation)) {
sameWeight = false; // 计算所有权重是否一样
}
}
if (totalWeight > 0 && ! sameWeight) {
// 如果权重不相同且权重大于0则按总权重数随机
int offset = random.nextInt(totalWeight);
// 并确定随机值落在哪个片断上
for (int i = 0; i < length; i++) {
offset -= getWeight(invokers.get(i), invocation);
if (offset < 0) {
return invokers.get(i);
}
}
}
// 如果权重相同或权重为0则均等随机
return invokers.get(random.nextInt(length));
}
这里我们通过负载均衡策略选择了一台Invoker返回给FailoverClusterInvoker.class，然后执行真正的invoker.
到此为止，我们关于dubbo的集群容错分析完毕了。这里只是分析了大致流程，后面会详细分析一下，集群容错机制和负载均衡相关内容。
总结
主要有三个流程：
1.在Directory选择出当前集群所有的invoker（List<Invoker>）
2.Router路由选择出符合规则的invoker(List<Invoker>)
3.选择合适的集群容错机制下，LoadBalance中根据负载均衡策略选择出一台可用的Invoker.
网友评论