Dubbo 的集群容错模式：Forking Cluster

本文主要是介绍Dubbo 的集群容错模式：Forking Cluster，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

集群容错系列文章：
Failover Cluster 失败自动切换
Failfast Cluster 快速失败，抛出异常
Failsafe Cluster 快速失败，不抛出异常
Failback Cluster 失败后定时重试
Forking Cluster 并行调用多个实例，只要一个成功就返回
Broadcast Cluster 广播调用所有实例，有一个报错则抛出异常
Available Cluster 可用的实例
Mergeable Cluster 合并结果

本文简单介绍 Dubbo 中的 Forking Cluster（并行调用多个服务器，只要一个成功就返回）。

简介

并行调用多个实例，只要一个成功即返回。通常用于实时性要求较高的读操作，但需要浪费更多服务资源。可通过 forks=”2” 来设置最大并行数。通过 timeout=”1000” 来设置调用超时时间。

如何使用

<dubbo:service cluster="forking" forks="2" timeout="1000" />

或

<dubbo:reference cluster="forking" forks="2" timeout="1000" />

实现逻辑

计算目前需要的并发数，通过负载均衡算法选中被调用实例列表
并发地调用实例列表，并将处理结果成功的放到阻塞队列中
获取处理结果队列中的第一个结果，判断是否是异常，是异常则抛出，不是异常则返回结果

源代码

public class ForkingClusterInvoker<T> extends AbstractClusterInvoker<T> {/*** Use {@link NamedInternalThreadFactory} to produce {@link org.apache.dubbo.common.threadlocal.InternalThread}* which with the use of {@link org.apache.dubbo.common.threadlocal.InternalThreadLocal} in {@link RpcContext}.*/// 使用 Cached 线程池private final ExecutorService executor = Executors.newCachedThreadPool(new NamedInternalThreadFactory("forking-cluster-timer", true));public ForkingClusterInvoker(Directory<T> directory) {super(directory);}@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;// 并行数量，默认是 2final int forks = getUrl().getParameter(Constants.FORKS_KEY, Constants.DEFAULT_FORKS);// 调用超时， 默认是 1sfinal int timeout = getUrl().getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);if (forks <= 0 || forks >= invokers.size()) {// 当并行数配置超过调用实例数量，则默认为调用实例数selected = invokers;} else {selected = new ArrayList<Invoker<T>>();for (int i = 0; i < forks; i++) {// TODO. Add some comment here, refer chinese version for more details.// 通过负载均衡算法选中实例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() {@Overridepublic 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(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();}}
}