集群调用
为了提供性能与保持高可用，通常一个接口能力会有多个提供者，也就是集群（Cluster）。
下面分析一下，Dubbo中一次invoke（调用）的流程。
首先介绍一下几个关键类，可以结合具体流程在进行消化。
cluster
各节点关系（参考Dubbo官网）
-
Cluster将Directory中的多个Invoker伪装成一个Invoker，对上层透明，伪装过程包含了容错逻辑，调用失败后，重试另一个
-
Invoker是Provider的一个可调用Service的抽象，Invoker封装了Provider地址及Service接口信息。
image
-
Directory代表多个Invoker，可以把它看成List<Invoker>，但与List不同的是，它的值可能是动态变化的，比如注册中心推送变更 -
Router负责从多个Invoker中按路由规则选出子集，比如读写分离，应用隔离等 -
LoadBalance负责从多个Invoker中选出具体的一个用于本次调用，选的过程包含了负载均衡算法，调用失败后，需要重选
invoke流程
使用的模板方法设计模式，可以看到一些主流程都在父类的AbstractClusterInvoker中，一些特殊实现如doXxx（比如doList、doInvoke）根据配置的不同子类，或者根据SPI定制化实现。
AbstractClusterInvoker#invoke
public Result invoke(final Invocation invocation) throws RpcException {
//校验服务可用性，是否已经被摧毁
checkWhetherDestroyed();
//将一些上下文信息绑定到invocation中，可以传递到远程服务器
Map<String, Object> contextAttachments = RpcContext.getContext().getObjectAttachments();
if (contextAttachments != null && contextAttachments.size() != 0) {
((RpcInvocation) invocation).addObjectAttachments(contextAttachments);
}
//将可用的invokers全部列出来（DynamicDirectory、RegistryDirectory），再根据router规则进行过滤选出子集，如读写分离等
List<Invoker<T>> invokers = list(invocation);
//进行负载均衡，选择真正调用的一个提供者，默认使用random
LoadBalance loadbalance = initLoadBalance(invokers, invocation);
//若是异步调用则记录调用id，为了幂等
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
//根据集群容错策略进行实际调用根据，默认failover
return doInvoke(invocation, invokers, loadbalance);
}
可以看到很多使用默认值，默认实现是如何选择的呢？可以看到是通过SPI机制指定缺省实现的
@SPI(Cluster.DEFAULT)
public interface Cluster {
String DEFAULT = FailoverCluster.NAME;
}
public class FailoverCluster extends AbstractCluster {
public final static String NAME = "failover";
}
集群容错
容错策略有很多，这里分析缺省实现Failover
Failover：当调用失败时，记录初始错误并重试其他调用者（重试n次，这意味着最多将调用n个不同的调用者）请注意，重试会导致延迟。默认延迟为5s。适合实时性不高的读操作。
FailoverClusterInvoker#doInvoke也就是父类最后一步的具体调用实现。
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
List<Invoker<T>> copyInvokers = invokers;
//校验invokers是否为空
checkInvokers(copyInvokers, invocation);
//获取方法名
String methodName = RpcUtils.getMethodName(invocation);
//从url中获取配置的重试次数并进行+1，若配置重试次数为2，那么最多调用3次。
int len = getUrl().getMethodParameter(methodName, RETRIES_KEY, DEFAULT_RETRIES) + 1;
//防御性编程，若配置的为负数或0也会进行一次调用
if (len <= 0) {
len = 1;
}
// retry loop.
// last exception.
RpcException le = null;
//已经调用过的会在后续负载均衡进行过滤。
List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyInvokers.size());
//调用过的服务进行地址记录
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已经被调用过则reselect一把
Invoker<T> invoker = select(loadbalance, invocation, copyInvokers, invoked);
//将调用过的放入list，日志记录使用
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 " + 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()) {
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);
}
整体流程
image
