Dubbo源码分析（八）集群容错机制
在上一章节，我们曾提到这样一个问题： 当调用服务失败后，我们怎么处理当前的请求？抛出异常亦或是重试？
为了解决这个问题，Dubbo 定义了集群接口 Cluster 以及 Cluster Invoker。集群 Cluster 用途是将多个服务提供者合并为一个 Cluster Invoker，并将这个 Invoker 暴露给服务消费者。这样一来，服务消费者只需通过这个 Invoker 进行远程调用即可，至于具体调用哪个服务提供者，以及调用失败后如何处理等问题，现在都交给集群模块去处理。
一、合并
在服务引用的过程中，我们最终会将一个或多个服务提供者Invoker封装成服务目录对象，但最后还要将它合并转换成Cluster Invoker对象。 Invoker invoker = cluster.join(directory);
这里的cluster就是扩展点自适应类，在Dubbo中默认是Failover，所以上面代码会调用到：
public class FailoverCluster implements Cluster {
public final static String NAME = "failover";
public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
return new FailoverClusterInvoker<T>(directory);
}
}
复制代码
上面的代码很简单，所以最后的Invoker对象指向的是 FailoverClusterInvoker
实例。它也是一个Invoker，它继承了抽象的 AbstractClusterInvoker
。
我们看下 AbstractClusterInvoker
类中的invoke方法。
public abstract class AbstractClusterInvoker<T> implements Invoker<T> {
public Result invoke(final Invocation invocation) throws RpcException {
LoadBalance loadbalance = null;
//调用服务目录，获取所有的服务提供者Invoker对象
List<Invoker<T>> invokers = directory.list(invocation);
if (invokers != null && !invokers.isEmpty()) {
//加载负载均衡组件
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).
getExtension(invokers.get(0).getUrl().
getMethodParameter(invocation.getMethodName(), "loadbalance", "random"));
}
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
//调用子类实现 ，不同的集群容错机制
return doInvoke(invocation, invokers, loadbalance);
}
}
复制代码
以上代码也很简单，我们分为三个步骤来看
- 调用服务目录，获取所有的服务提供者列表
- 加载负载均衡组件
- 调用子类实现，转发请求
关于负载均衡我们后续再深入了解，这是只知道它负责从多个Invoker中选取一个返回就行。
二、集群容错策略
Dubbo为我们提供了多种集群容错机制。主要如下：
- Failover Cluster - 失败自动切换
FailoverClusterInvoker在调用失败时，会自动切换 Invoker 进行重试。默认配置下，Dubbo 会使用这个类作为缺省 Cluster Invoker。
- Failfast Cluster - 快速失败
FailfastClusterInvoker 只会进行一次调用，失败后立即抛出异常。
- Failsafe Cluster - 失败安全
FailsafeClusterInvoker 当调用过程中出现异常时，仅会打印异常，而不会抛出异常。
- Failback Cluster - 失败自动恢复
FailbackClusterInvoker 会在调用失败后，返回一个空结果给服务提供者。并通过定时任务对失败的调用进行重传，适合执行消息通知等操作。
- Forking Cluster - 并行调用多个服务提供者
ForkingClusterInvoker 会在运行时通过线程池创建多个线程，并发调用多个服务提供者。只要有一个服务提供者成功返回了结果，doInvoke 方法就会立即结束运行。ForkingClusterInvoker 的应用场景是在一些对实时性要求比较高读操作（注意是读操作，并行写操作可能不安全）下使用，但这将会耗费更多的资源。
- BroadcastClusterInvoker - 广播
BroadcastClusterInvoker 会逐个调用每个服务提供者，如果其中一台报错，在循环调用结束后，BroadcastClusterInvoker 会抛出异常。该类通常用于通知所有提供者更新缓存或日志等本地资源信息。
三、自动切换
FailoverClusterInvoker 在调用失败时，会自动切换 Invoker 进行重试。我们重点看它的 doInvoke
方法。
public Result doInvoke(Invocation invocation,
final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
List<Invoker<T>> copyinvokers = invokers;
//检查invokers是否为空
checkInvokers(copyinvokers, invocation);
//获取重试次数 这里默认是3次
int len = getUrl().getMethodParameter(invocation.getMethodName(), "retries",2) + 1;
if (len <= 0) {
len = 1;
}
//异常信息对象
RpcException le = null; // last exception.
List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size());
Set<String> providers = new HashSet<String>(len);
//循环调用 失败重试len次
for (int i = 0; i < len; i++) {
if (i > 0) {
checkWhetherDestroyed();
//重新获取服务提供者列表
copyinvokers = list(invocation);
//再次检查
checkInvokers(copyinvokers, invocation);
}
//通过loadbalance选取一个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()) {
throw e;
}
le = e;
} catch (Throwable e) {
le = new RpcException(e.getMessage(), e);
} finally {
providers.add(invoker.getUrl().getAddress());
}
}
//重试失败
throw new RpcException("");
}
复制代码
我们可以看到，它的重点是invoker的调用是在一个循环方法中。只要不return，就会一直调用，重试 len 次。我们总结下它的过程：
- 检查invokers是否为空
- 获取重试次数，默认为3
- 进入循环
- 如果是重试，再次获取服务提供者列表，并校验
- 选取Invoker，并调用
- 无异常，返回结果，循环结束
- 捕获到异常，继续循环调用直至重试最大次数
四、快速失败
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()) {
throw (RpcException) e;
}
throw new RpcException("....");
}
}
复制代码
五、失败安全
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();
}
}
复制代码
六、自动恢复
FailbackClusterInvoker 会在调用失败后，也是打印异常信息并返回一个空的结果对象，但是还没结束，它还会偷偷开启一个定时任务，再次去调用。
protected 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("Failback to invoke method " + invocation.getMethodName() + ",
wait for retry in background. Ignored exception: "
+ e.getMessage() + ", ", e);
//添加失败信息
addFailed(invocation, this);
return new RpcResult();
}
}
复制代码
我们可以看到，调用失败后，除了打印异常信息和返回空结果对象之外，还有一个方法 addFailed
它就是开启定时任务的地方。
1、开启定时任务
首先，定义一个包含2个线程的线程池对象。
Executors.newScheduledThreadPool(2, new NamedThreadFactory("failback-cluster-timer", true));
然后，延迟5秒后，每隔5秒调用 retryFailed
方法，直到调用成功。
private void addFailed(Invocation invocation, AbstractClusterInvoker<?> router) {
if (retryFuture == null) {
synchronized (this) {
if (retryFuture == null) {
retryFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
public void run() {
try {
//重试方法
retryFailed();
} catch (Throwable t) {
logger.error("Unexpected error occur at collect statistic", t);
}
}
}, 5000, 5000, TimeUnit.MILLISECONDS);
}
}
}
//ConcurrentHashMap 添加失败任务
failed.put(invocation, router);
}
复制代码
最后，我们需要注意 failed.put(invocation, router);
它将当前失败的任务添加到failed，它是一个ConcurrentHashMap对象。
2、重试
重试的逻辑也不复杂，从failed对象中获取失败的记录，调用即可。
void retryFailed() {
//如果为空，说明已经没有了失败的任务
if (failed.size() == 0) {
return;
}
//遍历failed，对失败的调用进行重试
Set<Entry<Invocation, AbstractClusterInvoker<?>>> failedSet = failed.entrySet();
for (Entry<Invocation, AbstractClusterInvoker<?>> entry : failedSet) {
Invocation invocation = entry.getKey();
Invoker<?> invoker = entry.getValue();
try {
// 再次进行调用
invoker.invoke(invocation);
// 调用成功后，从 failed 中移除 invoker
failed.remove(invocation);
} catch (Throwable e) {
logger.error("......", e);
}
}
}
复制代码
如上代码，其中的重点是调用成功后，要将invocation移除。当再次调用到这个方法，开头的条件判断成立，就直接返回，不再继续调用。
七、并行调用
ForkingClusterInvoker 会在运行时通过线程池创建多个线程，并发调用多个服务提供者。只要有一个服务提供者成功返回了结果，doInvoke 方法就会立即结束运行。
public Result doInvoke(final Invocation invocation,
List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
final List<Invoker<T>> selected;
//获取最大并行数 默认为2
final int forks = getUrl().getParameter("forks", 2);
//超时时间
final int timeout = getUrl().getParameter("timeout", 1000);
if (forks <= 0 || forks >= invokers.size()) {
selected = invokers;
} else {
selected = new ArrayList<Invoker<T>>();
//选择Invoker 并添加到selected
for (int i = 0; i < forks; i++) {
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() {
@Override
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("....");
}
return (Result) ret;
} catch (InterruptedException e) {
throw new RpcException(e.getMessage(), e);
}
}
复制代码
以上代码的重点就是阻塞队列LinkedBlockingQueue。如果有结果放入，poll方法会立即返回，完成整个调用。我们再总结下整体流程：
- 获取最大并行数，默认为2；获取超时时间
- 选择Invoker，并添加到selected
- 通过newCachedThreadPool创建多个线程，调用服务。
- 正常返回后，将结果offer到队列。此时调用流程结束，返回正常信息。
- 调用服务异常后，判断异常次数是否大于等于最大并行数，条件成立则将异常信息offer到队列，此时调用流程结束，返回异常信息。
八、广播
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;
}
复制代码
正文到此结束
- 本文标签： 代码 key http IO 并发 负载均衡 IDE 实例 自适应 executor retry https 配置 Select provider cache 缓存 db map DOM dubbo Service synchronized 目录 安全 遍历 总结 Atom ArrayList 集群 CTO final queue ConcurrentHashMap rand cat CEO list message HashMap tk 线程 时间 value 线程池 id HashSet
- 版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
- 本文海报： 生成海报一 生成海报二
热门推荐
相关文章
近期评论
-
文章和留言都翻到11页了 没有OOM
-
-
朋友，翻页到11页，及以后，会出现OOM，无法访问
-
-
搞个gitee的项目
-
-
版本号是多少，你可以下载哪个代码仓库，jdk选1.8 直接跑就行
-
-
-
Loading...
![[HBLOG]公众号](https://www.liuhaihua.cn/img/qrcode_gzh.jpg)
