Dubbo源码分析（五）|容错策略

一、dubbo容错

1、配置方式

1）服务端设置
  <dubbo:service cluster="failsafe"  retries="2"/>
  
2）调用端设置
  <dubbo:reference cluster="failsafe"  retries="2"/>

2、FailoverClusterInvoker

FailoverClusterInvoker是一种失败后，重新换其他机器重试，并设有重试次数的一种容错机制

@Override
@SuppressWarnings({"unchecked", "rawtypes"})
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
    List<Invoker<T>> copyinvokers = invokers;
    //  对 copyinvokers 进行判空检查
    checkInvokers(copyinvokers, invocation);
    //获取重试次数，如果参数没配，默认是2次，再加上第一次调用，总共调用3次
    int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
    if (len <= 0) {
        len = 1;
    }
    // retry loop.
    RpcException le = null; // last exception.
    //可用服务端invoker，从这里面获取对应的invoker
    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++) {
        //Reselect before retry to avoid a change of candidate `invokers`.
        //NOTE: if `invokers` changed, then `invoked` also lose accuracy.
        if (i > 0) {
            //检查是否已经是被destroyed,如果被destroyed,则报错，停止循环调用
            checkWhetherDestroyed();
            //重新加载可用invoker列表
            copyinvokers = list(invocation);
            // check again 对 copyinvokers 进行判空检查
            checkInvokers(copyinvokers, invocation);
        }
        // 通过负载均衡选择 Invoker
        Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
        // 添加到 invoker 到 invoked 列表中，用于之后当第一次调用没成功的时候，用于排除这个服务器，调用其他服务器
        invoked.add(invoker);
        // 设置 invoked 到 RPC 上下文中
        RpcContext.getContext().setInvokers((List) invoked);
        try {
            // 调用目标 Invoker 的 invoke 方法
            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);
}  

3、FailfastClusterInvoker

FailfastClusterInvoker是一种快速失败的容错机制，也就是只会调用一次，之后没成功就抛出异常

public FailfastClusterInvoker(Directory<T> directory) {
    super(directory);
}

@Override
public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
    checkInvokers(invokers, invocation);
    // 通过负载均衡选择 Invoker
    Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
    try {
        // 调用 Invoker
        return invoker.invoke(invocation);
    } catch (Throwable e) {
        // 调用失败直接抛出异常
        if (e instanceof RpcException && ((RpcException) e).isBiz()) { // biz exception.
            throw (RpcException) e;
        }
        // 抛出异常
        throw new RpcException(e instanceof RpcException ? ((RpcException) e).getCode() : 0, "Failfast invoke providers " + invoker.getUrl() + " " + loadbalance.getClass().getSimpleName() + " select from all providers " + invokers + " for service " + getInterface().getName() + " method " + invocation.getMethodName() + " on consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e.getCause() != null ? e.getCause() : e);
    }
}

4、FailsafeClusterInvoker

failsafeClusterInvoker是一种安全失败的模式，即如果调用失败，不抛出异常，只是记录异常到日志，之后悄悄的返回return new RpcResult();空的RpcResult对象

@Override
public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
    try {
        //检查Invokerlist是不是为空 
        checkInvokers(invokers, invocation);
        // 通过负载均衡选择 Invoker
        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(); // ignore
    }
}

5.FailbackClusterInvoker

FailbackClusterInvoker是一种失败之后先返回，然后通过定时任务反复调用，直至成功

//将失败的调用放入到failed中，进行定时任务调用，成功就移除
private void addFailed(Invocation invocation, AbstractClusterInvoker<?> router) {
    if (retryFuture == null) {
        synchronized (this) {
            if (retryFuture == null) {
                // 创建定时任务，每隔5秒执行一次
                retryFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {

                    @Override
                    public void run() {
                        // collect retry statistics
                        try {
                            // 对失败的调用进行重试
                            retryFailed();
                        } catch (Throwable t) { // Defensive fault tolerance
                            // 如果发生异常，仅打印异常日志，不抛出
                            logger.error("Unexpected error occur at collect statistic", t);
                        }
                    }
                }, RETRY_FAILED_PERIOD, RETRY_FAILED_PERIOD, TimeUnit.MILLISECONDS);
            }
        }
    }
    // 添加 invocation 和 invoker 到 failed 中
    failed.put(invocation, router);
}

//该方法会对 failed 进行遍历，然后依次对 Invoker 进行调用。调用成功则将 Invoker 从 failed 中移除，调用失败则忽略失败原因
void retryFailed() {
    if (failed.size() == 0) {
        return;
    }
    // 遍历 failed，对失败的调用进行重试
    for (Map.Entry<Invocation, AbstractClusterInvoker<?>> entry : new HashMap<Invocation, AbstractClusterInvoker<?>>(
            failed).entrySet()) {
        Invocation invocation = entry.getKey();
        Invoker<?> invoker = entry.getValue();
        try {
            // 再次进行调用
            invoker.invoke(invocation);
            // 调用成功后，从 failed 中移除 invoker
            failed.remove(invocation);
        } catch (Throwable e) {
            logger.error("Failed retry to invoke method " + invocation.getMethodName() + ", waiting again.", e);
        }
    }
}

@Override
protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
    try {
        //检查invoker list是否为空
        checkInvokers(invokers, invocation);
        // 通过负载均衡选择 Invoker
        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(); // ignore
    }
}

6.ForkingClusterInvoker

ForkingClusterInvoker会在运行时通过线程池创建多个线程，并行调用多个服务提供者。只要有一个服务提供者调用成功，doInvoker方法立即结束。
ForkingClusterInvoker 的应用场景是在一些对实时性要求比较高读操作（注意是读操作，并行写操作可能不安全）下使用，但这将会耗费更多的资源

@Override
@SuppressWarnings({"unchecked", "rawtypes"})
public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
    //检查服务端可调用列表是否为空
    checkInvokers(invokers, invocation);
    final List<Invoker<T>> selected;
    // 获取 forks 配置forks，默认是2
    final int forks = getUrl().getParameter(Constants.FORKS_KEY, Constants.DEFAULT_FORKS);
    // 获取超时配置timeout 默认1000毫秒
    final int timeout = getUrl().getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
    // 如果 forks 配置不合理，则直接将 invokers 赋值给 selected
    if (forks <= 0 || forks >= invokers.size()) {
        selected = invokers;
    } else {
        selected = new ArrayList<Invoker<T>>();
        // 循环选出 forks 个 Invoker，并添加到 selected 中，这里的意思就是根据负载均衡算法生成多个Invoker，可能会重复
        for (int i = 0; i < forks; i++) {
            // 选择 Invoker
            Invoker<T> invoker = select(loadbalance, invocation, invokers, selected);
            if (!selected.contains(invoker)) {
                //将每次返回的invoker，添加到selected列表中
                selected.add(invoker);
            }
        }
    }

    // ----------------------分割线1 ---------------------- //
    RpcContext.getContext().setInvokers((List) selected);
    final AtomicInteger count = new AtomicInteger();
    final BlockingQueue<Object> ref = new LinkedBlockingQueue<Object>();
    // 遍历 selected 列表
    for (final Invoker<T> invoker : selected) {
        // 为每个 Invoker 创建一个执行线程
        executor.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    // 进行远程调用
                    Result result = invoker.invoke(invocation);
                    // 将结果存到阻塞队列中
                    ref.offer(result);
                } catch (Throwable e) {
                    // 仅在 value 大于等于 selected.size() 时，才将异常对象
                    // 放入阻塞队列中，在并行调用多个服务提供者的情况下，只要有一个服务提供者能够成功返回结果，而其他全部失败。
                    // 此时 ForkingClusterInvoker 仍应该返回成功的结果，而非抛出异常。
                    // 在value >= selected.size()时将异常对象放入阻塞队列中，可以保证异常对象不会出现在正常结果的前面，这样可从阻塞队列中优先取出正常的结果。
                    int value = count.incrementAndGet();
                    if (value >= selected.size()) {
                        // 将异常对象存入到阻塞队列中
                        ref.offer(e);
                    }
                }
            }
        });
    }

    // ---------------------分割线2--------------------- //

    try {
        // 从阻塞队列中取出远程调用结果,只要有一个正常结果返回了 就可以直接返回正常结果了。
        //比如，有4个服务端调用，启动了4个线程，然后只要有一个正常结果，就会放入到ref中，比如说线程1返回了正常结果，就可以正常返回了
        //但是有一种情况就是，4个线程返回的都是异常，但是只有当第4个线程回来的时候才能将异常放入到ref中，这个时候ref才有值，也就是说只有所有的线程都失败，才是真失败
        //那也就是说，4个里面只要又一个成功，那就是成功了
        Object ret = ref.poll(timeout, TimeUnit.MILLISECONDS);
        // 如果结果类型为 Throwable，则抛出异常
        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);
    }
}

7、BroadcastClusterInvoker

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;
    // 遍历 Invoker 列表，逐个调用
    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);
        }
    }
    // exception 不为空，则抛出异常
    if (exception != null) {
        throw exception;
    }
    return result;
}