zoukankan html css js c++ java

Dubbo 系列（07-4）集群容错

BDubbo 系列（07-4）集群容错 - 集群

BDubbo 系列（07-4）集群容错 - 集群

Spring Cloud Alibaba 系列目录 - Dubbo 篇

1. 背景介绍

相关文档推荐：

在 Dubbo 的整个集群容错流程中，首先经过 Directory 获取所有的 Invoker 列表，然后经过 Routers 根据路由规则过滤 Invoker，最后幸存下来的 Invoker 还需要经过负载均衡 LoadBalance 这一关，选出最终调用的 Invoker。在前篇文章已经分析了服务字典、服务路由、负载均衡的基本原理，接下来继续分析集群容错的整个流程 Cluster。

如果有多个 Invoker，消费者调用那个 Invoker 呢？如果调用失败怎么处理，是重试，还是抛出异常，亦或是只打印异常等？为了处理这些问题，Dubbo 定义了集群接口 Cluster 以及 Cluster Invoker。集群 Cluster 用途是将多个服务提供者合并为一个 Cluster Invoker，并将这个 Invoker 暴露给服务消费者。这样一来，服务消费者只需通过这个 Invoker 进行远程调用即可，至于具体调用哪个服务提供者，以及调用失败后如何处理等问题，现在都交给集群模块去处理。集群模块是服务提供者和服务消费者的中间层，为服务消费者屏蔽了服务提供者的情况，这样服务消费者就可以专心处理远程调用相关事宜。比如发请求，接受服务提供者返回的数据等。这就是集群的作用。

Cluster 本质是将多个 Invokers 包装成一个 Invoker，对消费者屏蔽内部的负载均衡和异常处理，这样消费者根本不用感知集群的内部信息。

图1 Dubbo集群层次

graph LR MockInvoker -.-> Invoker Invoker -.-> ZKRegister-ClusterInvoker Invoker -.-> NacosRegister-ClusterInvoker Invoker -.-> Dubbo-Invoker Invoker -.-> Rest-Invoker ZKRegister-ClusterInvoker -. group1 .-> group1-ClusterInvoker ZKRegister-ClusterInvoker -. group2 .-> group2-ClusterInvoker group1-ClusterInvoker -.-> Invoker1 group1-ClusterInvoker -.-> Invoker2 group2-ClusterInvoker -.-> ...1 NacosRegister-ClusterInvoker -.-> ...2 Dubbo-Invoker -.-> ...3 Rest-Invoker -.-> ...4

从上图也可以看出 Invoker 实体域是 Dubbo 的核心模型，整个集群容错都围绕 Invoker 展开。

1.1 集群容错

在对集群相关代码进行分析之前，这里有必要先来介绍一下集群容错的所有组件。包含 Cluster、Cluster Invoker、Directory、Router 和 LoadBalance 等。

图2 Dubbo集群容错组件

集群工作过程可分为两个阶段：

第一个阶段是在服务消费者初始化期间，集群 Cluster 实现类为服务消费者创建 Cluster Invoker 实例，即上图中的 merge 操作。
第二个阶段是在服务消费者进行远程调用时。以 FailoverClusterInvoker 为例，该类型 Cluster Invoker 首先会调用 Directory 的 list 方法列举 Invoker 列表（可将 Invoker 简单理解为服务提供者）。Directory 的用途是保存 Invoker，可简单类比为 List<Invoker>。其实现类 RegistryDirectory 是一个动态服务目录，可感知注册中心配置的变化，它所持有的 Invoker 列表会随着注册中心内容的变化而变化。每次变化后，RegistryDirectory 会动态增删 Invoker，并调用 Router 的 route 方法进行路由，过滤掉不符合路由规则的 Invoker。当 FailoverClusterInvoker 拿到 Directory 返回的 Invoker 列表后，它会通过 LoadBalance 从 Invoker 列表中选择一个 Invoker。最后 FailoverClusterInvoker 会将参数传给 LoadBalance 选择出的 Invoker 实例的 invoker 方法，进行真正的远程调用。

1.2 容错策略

Dubbo 提供了 9 种集群容错的实现。

表1 Dubbo集群9种容错策略

容错机制	说明
Failover Cluster	失败自动切换。Dubbo默认容错机制，会做负载均衡，自动切换其它服务器重试3次(默认次数)。使用场景：读或幂等写操作，重试会加大对下游服务提供者的压力。
Failback Cluster	失败自动恢复。失败后记录到队列中，通过定时器重试，会做负载均衡。使用场景：异步或最终一致性的请求。
Failfast Cluster	快速失败。请求失败后返回异常，不进行重试，会做负载均衡。使用场景：非幂等性操作。
Failsafe Cluster	失败安全。请求失败后忽略异常，不进行重试，会做负载均衡。使用场景：不关心调用是否成功，eg：日志记录。
Forking Cluster	同时调用多个服务，只要有一个成功就返回。使用场景：对实时性要求高的请求。
Broadcast Cluster	广播多个服务，只要有一个失败就失败，不需要做负载均衡。使用场景：通常用于用户状态更新后广播。
Available Cluster	最简单的方式，不会做负载均衡，遍历所有的服务列表，找到每一个可用的服务就直接调用。如果没有可用的节点，则直接抛出异常。
Mock Cluster	广播调用所有可用的服务，任意一个节点报错则报错。
Mergeable Cluster	将多个节点请求得到的结果进行合并。

Cluster 使用参考 Dubbo 集群容错 - 实战。可在 <dubbo:service> <dubbo:reference> <dubbo:consumer> <dubbo:provider> 标签上设置 cluster 属性。如：

<dubbo:service cluster="failsafe" />

2. Cluster 结构

2.1 Cluster 继承体系

图3 Dubbo集群继承体系图

总结： 和服务路由接口一样，Cluster 也是 SPI 接口，它是一个工厂类，用于创建具体的 ClusterInvoker。上述类图只显示了部分 Cluster 的实现。Cluster 接口定义如下：

@SPI(FailoverCluster.NAME)
public interface Cluster {
    @Adaptive
    <T> Invoker<T> join(Directory<T> directory) throws RpcException;
}

Dubbo 默认的集群容错策略是 FailoverCluster，即故障转移策略，当一台服务器发生故障时，切换到另一台服务器进行重试，Dubbo 中默认重试 3 次。

public class FailbackCluster implements Cluster {
    @Override
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
        return new FailbackClusterInvoker<T>(directory);
    }
}

2.2 AbstractClusterInvoker

上图可以看到，集群容错的具体实现都继承了抽象类 AbstractClusterInvoker，这个类主要完成了两件事：

实现的 Invoker 接口，对 Invoker#invoke 方法做了通用的抽象实现。
实现了通用的负载均衡算法。

2.2.1 invoke 执行

我们猜测一下 Invoker 的执行流程，执行前肯定需要获取所有的服务列表 invokers，然后根据负载均衡算法获取具体执行的 Invoker，最后才执行，至于调用失败怎么处理，则是具体的子类来做集群容错。

@Override
public Result invoke(final Invocation invocation) throws RpcException {
    checkWhetherDestroyed();

    // 1. 绑定 attachments 到 invocation 中.
    Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
    if (contextAttachments != null && contextAttachments.size() != 0) {
        ((RpcInvocation) invocation).addAttachments(contextAttachments);
    }

    // 2. 通过 Directory 列举所有的 Invoker
    List<Invoker<T>> invokers = list(invocation);
    // 3. 加载 LoadBalance
    LoadBalance loadbalance = initLoadBalance(invokers, invocation);
    RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
    // 4. 调用 doInvoke 进行后续操作
    return doInvoke(invocation, invokers, loadbalance);
}

总结： 其实 Dubbo 的实现和猜测的差不多，这里还会绑定 attachments 参数。之后通过 Directory 获取所有的 invokers，初始化 loadbalance，具体的执行逻辑则是委托给子类实现。

注意： list(invocation) 方法调用 directory.list(invocation) 时已经经过路由规则过滤，此时只需要进行负载均衡算法即可。

2.2.2 负载均衡

AbstractClusterInvoker 并没有直接使用 Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation) 进行负载均衡，而是进一步做了封装。

如果开启了粘滞连接，则需要将上次使用的 Invoker 缓存起来，只要服务可用就直接调用，不再需要进行负载均衡。
如果调用失败，则需要重新进行负载均衡，此时需要排除已经重试过的服务。

图4 Dubbo集群负载均衡调用过程

graph LR AbstractClusterInvoker -- 粘滞连接 --> select select -- loadbalance.select --> doSelect doSelect -- 重试 --> reselect

总结： AbstractClusterInvoker 调用 select 进行负载均衡时

select 方法主要处理粘滞连接。
doSelect 方法调用负载均衡算法 loadbalance.select。
reselect 方法当 doSelect 选出的服务不可用时，则需要重试进行负载均衡。

（1）粘滞连接

select 方法主要处理粘滞连接。select 方法有四个参数：第一个参数为负载均衡算法；第二个为调用参数；第三个为所有注册的服务列表，第四个为已经重试过后服务。

protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation,
		List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {

    if (CollectionUtils.isEmpty(invokers)) {
        return null;
    }
    // 1. 获取调用方法名
    String methodName = invocation == null ? StringUtils.EMPTY : invocation.getMethodName();

    // 2. 获取 sticky 配置，sticky 表示粘滞连接。所谓粘滞连接是指让服务消费者尽可能的
    //    调用同一个服务提供者，除非该提供者挂了再进行切换
    boolean sticky = invokers.get(0).getUrl()
        .getMethodParameter(methodName, CLUSTER_STICKY_KEY, DEFAULT_CLUSTER_STICKY);

    // 3. 检测 invokers 列表是否包含 stickyInvoker，如果不包含，
    //    说明 stickyInvoker 代表的服务提供者挂了，此时需要将其置空
    if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
        stickyInvoker = null;
    }

    // 4. 如果是粘滞连接，则需要判断这个服务是否已经重试过了，暂时不可用
    //    sticky && stickyInvoker != null 表示是粘滞连接
    //    (selected == null || !selected.contains(stickyInvoker)) 表示服务未重试
    if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
        // availablecheck=true 表示每次都要判断服务是否可用
        if (availablecheck && stickyInvoker.isAvailable()) {
            return stickyInvoker;
        }
    }

    // 5. 如果线程走到当前代码处，说明前面的 stickyInvoker 为空，或者不可用。
    //    此时继续调用 doSelect 选择 Invoker
    Invoker<T> invoker = doSelect(loadbalance, invocation, invokers, selected);

    // 6. sticky=true，则将负载均衡组件选出的 Invoker 缓存起来
    if (sticky) {
        stickyInvoker = invoker;
    }
    return invoker;
}

总结： 可以看到 select 主要在处理粘滞连接，如果开启了粘滞连接，且服务可用时直接返回这个 stickyInvoker。否则才调用 doSelect 进行负载均衡。

（2）负载均衡

private Invoker<T> doSelect(LoadBalance loadbalance, Invocation invocation,
		List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
    // 1. 判断是否需要进行负载均衡
    if (CollectionUtils.isEmpty(invokers)) {
        return null;
    }
    if (invokers.size() == 1) {
        return invokers.get(0);
    }

    // 2. 通过负载均衡组件选择 Invoker
    Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);

    // 3. 如果负载均衡选出的 Invoker 已经重试过了或不可用，则需要重选 reselect
    if ((selected != null && selected.contains(invoker))
        || (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
        try {
            // 3.1 进行重选
            Invoker<T> rInvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
            // 3.2 重选的 rInvoker 不为空，直接返回这个 rInvoker
            if (rInvoker != null) {
                invoker = rInvoker;
            // 3.3 rinvoker 为空，则返回下一个(相对于负载均衡选出的invoker)
            //     这也可以看成重选的部分逻辑
            } else {
                int index = invokers.indexOf(invoker);
                try {
                    invoker = invokers.get((index + 1) % invokers.size());
                } catch (Exception e) {
                    logger.warn(e.getMessage() + " may because invokers list dynamic change, ignore.", e);
                }
            }
        } catch (Throwable t) {
        }
    }
    return invoker;
}

总结： doSelect 主要做了两件事，第一是通过负载均衡组件选择 Invoker。第二是，如果选出来的 Invoker 不稳定，或不可用，此时需要调用 reselect 方法进行重选。若 reselect 选出来的 Invoker 为空，此时定位负载均衡选出的 invoker 在 invokers 列表中的位置 index，然后获取 index + 1 处的 invoker，这也可以看做是重选逻辑的一部分。下面我们来看一下 reselect 方法的逻辑。

（3）重选

reselect 重新进行负载均衡，首先对未重试的可用 invokers 进行负载均衡，如果已经全部重试过了，则将重试过的服务中过滤出可用的服务重新进行负载均衡。

private Invoker<T> reselect(LoadBalance loadbalance, Invocation invocation,
	List<Invoker<T>> invokers, List<Invoker<T>> selected,
	boolean availablecheck) throws RpcException {

    List<Invoker<T>> reselectInvokers = new ArrayList<>(
        invokers.size() > 1 ? (invokers.size() - 1) : invokers.size());

    // 1. 将不在 selected 集合中的 invokers 过滤出来进行负载均衡
    for (Invoker<T> invoker : invokers) {
        if (availablecheck && !invoker.isAvailable()) {
            continue;
        }

        if (selected == null || !selected.contains(invoker)) {
            reselectInvokers.add(invoker);
        }
    }

    // reselectInvokers 不为空，此时通过负载均衡组件进行选择
    if (!reselectInvokers.isEmpty()) {
        return loadbalance.select(reselectInvokers, getUrl(), invocation);
    }

    // 2. 只能对 selected(已经select过的) 中可用的 invoker 再次进行负载均衡
    if (selected != null) {
        for (Invoker<T> invoker : selected) {
            if ((invoker.isAvailable()) && !reselectInvokers.contains(invoker)) {
                reselectInvokers.add(invoker);
            }
        }
    }
    if (!reselectInvokers.isEmpty()) {
        return loadbalance.select(reselectInvokers, getUrl(), invocation);
    }

    return null;
}

总结： reselect 也做了容错处理，代码可分为两部分：

第一部分：在未重试过的服务中重新进行负载均衡。
第二部分：如果已经全部重试过了，则在重试过的服务中过滤可用的服务重新进行负载均衡。

3. 集群容错

3.1 FailoverClusterInvoker

FailoverClusterInvoker 故障转移，即在调用失败时，会自动切换 Invoker 进行重试。默认确配置下，Dubbo 会使用这个类作为缺省 Cluster Invoker。下面来看一下该类的逻辑。

public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers,
                       LoadBalance loadbalance) throws RpcException {
    List<Invoker<T>> copyInvokers = invokers;
    checkInvokers(copyInvokers, invocation);
    // 1. 参数获取，如重试次数
    String methodName = RpcUtils.getMethodName(invocation);
    int len = getUrl().getMethodParameter(methodName, RETRIES_KEY, DEFAULT_RETRIES) + 1;
    if (len <= 0) {
        len = 1;
    }
    RpcException le = null;
    List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyInvokers.size());
    Set<String> providers = new HashSet<String>(len);
    // 2. 循环调用，失败重试，默认为 3 次
    for (int i = 0; i < len; i++) {
        // 3. 第一次传进来的invokers已经check过了，第二次则是重试，需要重新获取最新的服务列表
        if (i > 0) {
            checkWhetherDestroyed();
            // 通过调用 list 可得到最新可用的 Invoker 列表，并check是否为空
            copyInvokers = list(invocation);
            checkInvokers(copyInvokers, invocation);
        }
        // 4. 核心代码：通过负载均衡选择 Invoker
        Invoker<T> invoker = select(loadbalance, invocation, copyInvokers, invoked);
        // 5. 已经重试边的添加到invoked列表中，下一次重试时就会过滤这个服务
        invoked.add(invoker);
        RpcContext.getContext().setInvokers((List) invoked);
        try {
            // 6. 核心代码：调用目标 Invoker 的 invoke 方法
            Result result = invoker.invoke(invocation);
            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());
        }
    }
    // 7. 若重试失败，则抛出异常
    throw new RpcException(le);
}

总结： 有了上面的基础，看 FailoverClusterInvoker 的代码应该很轻松了。只要执行失败就重新调用 select(loadbalance, invocation, copyInvokers, invoked) 进行重试，Dubbo 默认重试 3 次。

3.2 FailbackClusterInvoker

FailbackClusterInvoker 会在调用失败后，返回一个空结果给服务提供者。并通过定时任务对失败的调用进行重传，适合执行消息通知等操作。下面来看一下它的实现逻辑。

@Override
protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
    Invoker<T> invoker = null;
    try {
        checkInvokers(invokers, invocation);
        invoker = select(loadbalance, invocation, invokers, null);
        return invoker.invoke(invocation);
    } catch (Throwable e) {
        // 任务失败后添加定时器中重试
        addFailed(loadbalance, invocation, invokers, invoker);
        return AsyncRpcResult.newDefaultAsyncResult(null, null, invocation); // ignore
    }
}

总结： FailbackClusterInvoker 失败后添加到定时器重试，默认每隔 5s 执行一次，重试 3 次。在 Dubbo 系列（03）注册中心时也提到过这个定时器 HashedWheelTimer，当注册失败时也有失败重试的补偿机制 FailbackRegistry。

private void addFailed(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, Invoker<T> lastInvoker) {
    // 1. 初始化定时器
    if (failTimer == null) {
        synchronized (this) {
            if (failTimer == null) {
                failTimer = new HashedWheelTimer(
                    new NamedThreadFactory("failback-cluster-timer", true),
                    1, TimeUnit.SECONDS, 32, failbackTasks);
            }
        }
    }
    // 2. 重试任务添加到定时器中
    RetryTimerTask retryTimerTask = new RetryTimerTask(loadbalance, invocation, 
		invokers, lastInvoker, retries, RETRY_FAILED_PERIOD);
    try {
        failTimer.newTimeout(retryTimerTask, RETRY_FAILED_PERIOD, TimeUnit.SECONDS);
    } catch (Throwable e) {
    }
}

总结： addFailed 主要是将任务添加到定时器 HashedWheelTimer，默认 5s 执行一次，重试 3 次。具体任务执行在 RetryTimerTask#run 方法中。

@Override
public void run(Timeout timeout) {
    try {
        // 重新进行负载均衡，失败后又等 5s 再重试一次
        Invoker<T> retryInvoker = select(loadbalance, invocation, invokers,
			Collections.singletonList(lastInvoker));
        lastInvoker = retryInvoker;
        retryInvoker.invoke(invocation);
    } catch (Throwable e) {
        if ((++retryTimes) >= retries) {
        } else {
            rePut(timeout);
        }
    }
}

3.3 FailfastClusterInvoker

FailfastClusterInvoker 只会进行一次调用，失败后立即抛出异常。适用于幂等操作，比如新增记录。源码如下：

@Override
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(e);
    }
}

总结： 执行失败就抛出异常。

3.4 FailsafeClusterInvoker

FailsafeClusterInvoker 只会进行一次调用，失败后打印日志，返回 null。适用于写入审计日志等操作。

@Override
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 AsyncRpcResult.newDefaultAsyncResult(null, null, invocation); // ignore
    }
}

... 以后再研究。

每天用心记录一点点。内容也许不重要，但习惯很重要！

查看全文

相关阅读:
【pytest学习10】fixture参数化，fixture(params=data)装饰器的data是函数返回值yield request.param ，将带到下面调用为参数的函数中
 Pipfile 文件转换利器——pipfile-freeze
npm的lock
调试wmi连接主机进行监控
 RPC电源监控总结
 HTTP协议三次握手过程
 linux常用命令集
 Gym
Gym
实验2.2

原文地址：https://www.cnblogs.com/binarylei/p/11683295.html