37 集群容错:一个好汉三个帮(上)

你好,我是杨四正,今天我和你分享的主题是集群容错:一个好汉三个帮(上篇)。

在前面的课时中,我们已经对 Directory、Router、LoadBalance 等概念进行了深入的剖析,本课时将重点分析 Cluster 接口的相关内容。

Cluster 接口提供了我们常说的集群容错功能。

集群中的单个节点有一定概率出现一些问题,例如,磁盘损坏、系统崩溃等,导致节点无法对外提供服务,因此在分布式 RPC 框架中,必须要重视这种情况。为了避免单点故障,我们的 Provider 通常至少会部署在两台服务器上,以集群的形式对外提供服务,对于一些负载比较高的服务,则需要部署更多 Provider 来抗住流量。

在 Dubbo 中,通过 Cluster 这个接口把一组可供调用的 Provider 信息组合成为一个统一的 Invoker 供调用方进行调用。经过 Router 过滤、LoadBalance 选址之后,选中一个具体 Provider 进行调用,如果调用失败,则会按照集群的容错策略进行容错处理

Dubbo 默认内置了若干容错策略,并且每种容错策略都有自己独特的应用场景,我们可以通过配置选择不同的容错策略。如果这些内置容错策略不能满足需求,我们还可以通过自定义容错策略进行配置

了解了上述背景知识之后,下面我们就正式开始介绍 Cluster 接口。

Cluster 接口与容错机制

Cluster 的工作流程大致可以分为两步(如下图所示):①创建 Cluster Invoker 实例(在 Consumer 初始化时,Cluster 实现类会创建一个 Cluster Invoker 实例,即下图中的 merge 操作);②使用 Cluster Invoker 实例(在 Consumer 服务消费者发起远程调用请求的时候,Cluster Invoker 会依赖前面课时介绍的 Directory、Router、LoadBalance 等组件得到最终要调用的 Invoker 对象)。

Lark20201201-164714.png

Cluster 核心流程图

Cluster Invoker 获取 Invoker 的流程大致可描述为如下:

  1. 通过 Directory 获取 Invoker 列表,以 RegistryDirectory 为例,会感知注册中心的动态变化,实时获取当前 Provider 对应的 Invoker 集合;
  2. 调用 Router 的 route() 方法进行路由,过滤掉不符合路由规则的 Invoker 对象;
  3. 通过 LoadBalance 从 Invoker 列表中选择一个 Invoker;
  4. ClusterInvoker 会将参数传给 LoadBalance 选择出的 Invoker 实例的 invoke 方法,进行真正的远程调用。

这个过程是一个正常流程,没有涉及容错处理。Dubbo 中常见的容错方式有如下几个。

下面我们再来看 Cluster 接口。Cluster 接口是一个扩展接口,通过 @SPI 注解的参数我们知道其使用的默认实现是 FailoverCluster,它只定义了一个 join() 方法,在其上添加了 @Adaptive 注解,会动态生成适配器类,其中会优先根据 Directory.getUrl() 方法返回的 URL 中的 cluster 参数值选择扩展实现,若无 cluster 参数则使用默认的 FailoverCluster 实现。Cluster 接口的具体定义如下所示:

@SPI(FailoverCluster.NAME)

public interface Cluster {

    @Adaptive

    <T> Invoker<T> join(Directory<T> directory) throws RpcException;

}

Cluster 接口的实现类如下图所示,分别对应前面提到的多种容错策略:

Lark20201201-164718.png

Cluster 接口继承关系

在每个 Cluster 接口实现中,都会创建对应的 Invoker 对象,这些都继承自 AbstractClusterInvoker 抽象类,如下图所示:

Lark20201201-164728.png

AbstractClusterInvoker 继承关系图

通过上面两张继承关系图我们可以看出,Cluster 接口和 Invoker 接口都会有相应的抽象实现类,这些抽象实现类都实现了一些公共能力。下面我们就来深入介绍 AbstractClusterInvoker 和 AbstractCluster 这两个抽象类。

AbstractClusterInvoker

了解了 Cluster Invoker 的继承关系之后,我们首先来看 AbstractClusterInvoker,它有两点核心功能:一个是实现的 Invoker 接口,对 Invoker.invoke() 方法进行通用的抽象实现;另一个是实现通用的负载均衡算法。

在 AbstractClusterInvoker.invoke() 方法中,会通过 Directory 获取 Invoker 列表,然后通过 SPI 初始化 LoadBalance,最后调用 doInvoke() 方法执行子类的逻辑。在 Directory.list() 方法返回 Invoker 集合之前,已经使用 Router 进行了一次筛选,你可以回顾前面[第 31 课时]对 RegistryDirectory 的分析。

public Result invoke(final Invocation invocation) throws RpcException {

    // 检测当前Invoker是否已销毁

    checkWhetherDestroyed(); 

    // 将RpcContext中的attachment添加到Invocation中

    Map<String, Object> contextAttachments = RpcContext.getContext().getObjectAttachments();

    if (contextAttachments != null && contextAttachments.size() != 0) {

        ((RpcInvocation) invocation).addObjectAttachments(contextAttachments);

    }

    // 通过Directory获取Invoker对象列表,通过对RegistryDirectory的介绍我们知道,其中已经调用了Router进行过滤

    List<Invoker<T>> invokers = list(invocation);

    // 通过SPI加载LoadBalance

    LoadBalance loadbalance = initLoadBalance(invokers, invocation);

    RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);

    // 调用doInvoke()方法,该方法是个抽象方法

    return doInvoke(invocation, invokers, loadbalance);

}

protected List<Invoker<T>> list(Invocation invocation) throws RpcException {

    return directory.list(invocation); // 调用Directory.list()方法

}

下面我们来看一下 AbstractClusterInvoker 是如何按照不同的 LoadBalance 算法从 Invoker 集合中选取最终 Invoker 对象的。

AbstractClusterInvoker 并没有简单粗暴地使用 LoadBalance.select() 方法完成负载均衡,而是做了进一步的封装,具体实现在 select() 方法中。在 select() 方法中会根据配置决定是否开启粘滞连接特性,如果开启了,则需要将上次使用的 Invoker 缓存起来,只要 Provider 节点可用就直接调用,不会再进行负载均衡。如果调用失败,才会重新进行负载均衡,并且排除已经重试过的 Provider 节点。

// 第一个参数是此次使用的LoadBalance实现,第二个参数Invocation是此次服务调用的上下文信息,

// 第三个参数是待选择的Invoker集合,第四个参数用来记录负载均衡已经选出来、尝试过的Invoker集合

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

    if (CollectionUtils.isEmpty(invokers)) {

        return null;

    }

    // 获取调用方法名

    String methodName = invocation == null ? StringUtils.EMPTY_STRING : invocation.getMethodName();

    // 获取sticky配置,sticky表示粘滞连接,所谓粘滞连接是指Consumer会尽可能地

    // 调用同一个Provider节点,除非这个Provider无法提供服务

    boolean sticky = invokers.get(0).getUrl()

            .getMethodParameter(methodName, CLUSTER_STICKY_KEY, DEFAULT_CLUSTER_STICKY);

    // 检测invokers列表是否包含sticky Invoker,如果不包含,

    // 说明stickyInvoker代表的服务提供者挂了,此时需要将其置空

    if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {

        stickyInvoker = null;

    }

    // 如果开启了粘滞连接特性,需要先判断这个Provider节点是否已经重试过了

    if (sticky && stickyInvoker != null // 表示粘滞连接

            && (selected == null || !selected.contains(stickyInvoker)) // 表示stickyInvoker未重试过

    ) {

        // 检测当前stickyInvoker是否可用,如果可用,直接返回stickyInvoker

        if (availablecheck && stickyInvoker.isAvailable()) { 

            return stickyInvoker;

        }

    }

    // 执行到这里,说明前面的stickyInvoker为空,或者不可用

    // 这里会继续调用doSelect选择新的Invoker对象

    Invoker<T> invoker = doSelect(loadbalance, invocation, invokers, selected);

    if (sticky) { // 是否开启粘滞,更新stickyInvoker字段

        stickyInvoker = invoker;

    }

    return invoker;

}

doSelect() 方法主要做了两件事:

private Invoker<T> doSelect(LoadBalance loadbalance, Invocation invocation,

                            List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {

    // 判断是否需要进行负载均衡,Invoker集合为空,直接返回null

    if (CollectionUtils.isEmpty(invokers)) {

        return null;

    }

    if (invokers.size() == 1) { // 只有一个Invoker对象,直接返回即可

        return invokers.get(0);

    }

    // 通过LoadBalance实现选择Invoker对象

    Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);

    // 如果LoadBalance选出的Invoker对象,已经尝试过请求了或不可用,则需要调用reselect()方法重选

    if ((selected != null && selected.contains(invoker)) // Invoker已经尝试调用过了,但是失败了

            || (!invoker.isAvailable() && getUrl() != null && availablecheck) // Invoker不可用

    ) {

        try {

            // 调用reselect()方法重选

            Invoker<T> rInvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);

            // 如果重选的Invoker对象不为空,则直接返回这个 rInvoker

            if (rInvoker != null) {

                invoker = rInvoker;

            } else {

                int index = invokers.indexOf(invoker);

                try {

                    // 如果重选的Invoker对象为空,则返回该Invoker的下一个Invoker对象

                    invoker = invokers.get((index + 1) % invokers.size());

                } catch (Exception e) {

                    logger.warn("...");

                }

            }

        } catch (Throwable t) {

            logger.error("...");

        }

    }

    return invoker;

}

reselect() 方法会重新进行一次负载均衡,首先对未尝试过的可用 Invokers 进行负载均衡,如果已经全部重试过了,则将尝试过的 Provider 节点过滤掉,然后在可用的 Provider 节点中重新进行负载均衡。

private Invoker<T> reselect(LoadBalance loadbalance, Invocation invocation,

                            List<Invoker<T>> invokers, List<Invoker<T>> selected, boolean availablecheck) throws RpcException {

    // 用于记录要重新进行负载均衡的Invoker集合

    List<Invoker<T>> reselectInvokers = new ArrayList<>(

            invokers.size() > 1 ? (invokers.size() - 1) : invokers.size());

    // 将不在selected集合中的Invoker过滤出来进行负载均衡

    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);

    }

    // 只能对selected集合中可用的Invoker再次进行负载均衡

    if (selected != null) {

        for (Invoker<T> invoker : selected) {

            if ((invoker.isAvailable()) // available first

                    && !reselectInvokers.contains(invoker)) {

                reselectInvokers.add(invoker);

            }

        }

    }

    if (!reselectInvokers.isEmpty()) {

        return loadbalance.select(reselectInvokers, getUrl(), invocation);

    }

    return null;

}

AbstractCluster

常用的 ClusterInvoker 实现都继承了 AbstractClusterInvoker 类型,对应的 Cluster 扩展实现都继承了 AbstractCluster 抽象类。AbstractCluster 抽象类的核心逻辑是在 ClusterInvoker 外层包装一层 ClusterInterceptor,从而实现类似切面的效果

下面是 ClusterInterceptor 接口的定义:

@SPI

public interface ClusterInterceptor {

    // 前置拦截方法

    void before(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation);

    // 后置拦截方法

    void after(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation);

    // 调用ClusterInvoker的invoke()方法完成请求

    default Result intercept(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation) throws RpcException {

        return clusterInvoker.invoke(invocation);

    }

    // 这个Listener用来监听请求的正常结果以及异常

    interface Listener {

        void onMessage(Result appResponse, AbstractClusterInvoker<?> clusterInvoker, Invocation invocation);

        void onError(Throwable t, AbstractClusterInvoker<?> clusterInvoker, Invocation invocation);

    }

}

在 AbstractCluster 抽象类的 join() 方法中,首先会调用 doJoin() 方法获取最终要调用的 Invoker 对象,doJoin() 是个抽象方法,由 AbstractCluster 子类根据具体的策略进行实现。之后,AbstractCluster.join() 方法会调用 buildClusterInterceptors() 方法加载 ClusterInterceptor 扩展实现类,对 Invoker 对象进行包装。具体实现如下:

private <T> Invoker<T> buildClusterInterceptors(AbstractClusterInvoker<T> clusterInvoker, String key) {

    AbstractClusterInvoker<T> last = clusterInvoker;

    // 通过SPI方式加载ClusterInterceptor扩展实现

    List<ClusterInterceptor> interceptors = ExtensionLoader.getExtensionLoader(ClusterInterceptor.class).getActivateExtension(clusterInvoker.getUrl(), key);

    if (!interceptors.isEmpty()) {

        for (int i = interceptors.size() - 1; i >= 0; i--) {

            // 将InterceptorInvokerNode收尾连接到一起,形成调用链

            final ClusterInterceptor interceptor = interceptors.get(i);

            final AbstractClusterInvoker<T> next = last;

            last = new InterceptorInvokerNode<>(clusterInvoker, interceptor, next);

        }

    }

    return last;

}

@Override

public <T> Invoker<T> join(Directory<T> directory) throws RpcException {

    // 扩展名称由reference.interceptor参数确定

    return buildClusterInterceptors(doJoin(directory), directory.getUrl().getParameter(REFERENCE_INTERCEPTOR_KEY));

}

InterceptorInvokerNode 会将底层的 AbstractClusterInvoker 对象以及关联的 ClusterInterceptor 对象封装到一起,还会维护一个 next 引用,指向下一个 InterceptorInvokerNode 对象

在 InterceptorInvokerNode.invoke() 方法中,会先调用 ClusterInterceptor 的前置逻辑,然后执行 intercept() 方法调用 AbstractClusterInvoker 的 invoke() 方法完成远程调用,最后执行 ClusterInterceptor 的后置逻辑。具体实现如下:

public Result invoke(Invocation invocation) throws RpcException {

    Result asyncResult;

    try {

        interceptor.before(next, invocation); // 前置逻辑

        // 执行invoke()方法完成远程调用

        asyncResult = interceptor.intercept(next, invocation);

    } catch (Exception e) {

        if (interceptor instanceof ClusterInterceptor.Listener) {

            // 出现异常时,会触发监听器的onError()方法

            ClusterInterceptor.Listener listener = (ClusterInterceptor.Listener) interceptor;

            listener.onError(e, clusterInvoker, invocation);

        }

        throw e;

    } finally {

        // 执行后置逻辑

        interceptor.after(next, invocation);

    }

    return asyncResult.whenCompleteWithContext((r, t) -> {

        if (interceptor instanceof ClusterInterceptor.Listener) {

            ClusterInterceptor.Listener listener = (ClusterInterceptor.Listener) interceptor;

            if (t == null) {

                // 正常返回时,会调用onMessage()方法触发监听器

                listener.onMessage(r, clusterInvoker, invocation);

            } else {

                listener.onError(t, clusterInvoker, invocation);

            }

        }

    });

}

Dubbo 提供了两个 ClusterInterceptor 实现类,分别是 ConsumerContextClusterInterceptor 和 ZoneAwareClusterInterceptor,如下图所示:

Lark20201201-164721.png

ClusterInterceptor 继承关系图

在 ConsumerContextClusterInterceptor 的 before() 方法中,会在 RpcContext 中设置当前 Consumer 地址、此次调用的 Invoker 等信息,同时还会删除之前与当前线程绑定的 Server Context。在 after() 方法中,会删除本地 RpcContext 的信息。ConsumerContextClusterInterceptor 的具体实现如下:

public void before(AbstractClusterInvoker<?> invoker, Invocation invocation) {

    // 获取当前线程绑定的RpcContext

    RpcContext context = RpcContext.getContext();

    // 设置Invoker、Consumer地址等信息 context.setInvocation(invocation).setLocalAddress(NetUtils.getLocalHost(), 0);

    if (invocation instanceof RpcInvocation) {

        ((RpcInvocation) invocation).setInvoker(invoker);

    }

    RpcContext.removeServerContext();

}

public void after(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation) {

    RpcContext.removeContext(true); // 删除本地RpcContext的信息

}

ConsumerContextClusterInterceptor 同时继承了 ClusterInterceptor.Listener 接口,在其 onMessage() 方法中,会获取响应中的 attachments 并设置到 RpcContext 中的 SERVER_LOCAL 之中,具体实现如下:

public void onMessage(Result appResponse, AbstractClusterInvoker<?> invoker, Invocation invocation) {

// 从AppResponse中获取attachment,并设置到SERVER_LOCAL这个RpcContext中    RpcContext.getServerContext().setObjectAttachments(appResponse.getObjectAttachments());

}

介绍完 ConsumerContextClusterInterceptor,我们再来看 ZoneAwareClusterInterceptor。

在 ZoneAwareClusterInterceptor 的 before() 方法中,会从 RpcContext 中获取多注册中心相关的参数并设置到 Invocation 中(主要是 registry_zone 参数和 registry_zone_force 参数,这两个参数的具体含义,在后面分析 ZoneAwareClusterInvoker 时详细介绍),ZoneAwareClusterInterceptor 的 after() 方法为空实现。ZoneAwareClusterInterceptor 的具体实现如下:

public void before(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation) {

    RpcContext rpcContext = RpcContext.getContext();

    // 从RpcContext中获取registry_zone参数和registry_zone_force参数

    String zone = (String) rpcContext.getAttachment(REGISTRY_ZONE);

    String force = (String) rpcContext.getAttachment(REGISTRY_ZONE_FORCE);

    // 检测用户是否提供了ZoneDetector接口的扩展实现

    ExtensionLoader<ZoneDetector> loader = ExtensionLoader.getExtensionLoader(ZoneDetector.class);

    if (StringUtils.isEmpty(zone) && loader.hasExtension("default")) {

        ZoneDetector detector = loader.getExtension("default");

        zone = detector.getZoneOfCurrentRequest(invocation);

        force = detector.isZoneForcingEnabled(invocation, zone);

    }

    // 将registry_zone参数和registry_zone_force参数设置到Invocation中

    if (StringUtils.isNotEmpty(zone)) {

        invocation.setAttachment(REGISTRY_ZONE, zone);

    }

    if (StringUtils.isNotEmpty(force)) {

        invocation.setAttachment(REGISTRY_ZONE_FORCE, force);

    }

}

需要注意的是,ZoneAwareClusterInterceptor 没有实现 ClusterInterceptor.Listener 接口,也就是不提供监听响应的功能。

总结

本课时我们主要介绍的是 Dubbo Cluster 层中容错机制相关的内容。首先,我们了解了集群容错机制的作用。然后,我们介绍了 Cluster 接口的定义以及其各个实现类的核心功能。之后,我们深入讲解了 AbstractClusterInvoker 的实现,其核心是实现了一套通用的负载均衡算法。最后,我们还分析了 AbstractCluster 抽象实现类以及其中涉及的 ClusterInterceptor 接口的内容。