dubbo-------初识dubbo集群容错机制

Lord Lean Notes 2020-01-02

346

dubbo是一个远程RPC调用框架，我们在实际工作时可能会有多台被调用的服务器，那么我们怎么保证当一台服务器挂掉时，整个服务仍是可以运行的，这就需要集群容错机制来进行预防，现在就让我们来学习一下dubbo的几种集群机制。

dubbo一共有七种集群容错策略：

failover策略
broadcast策略
forking策略
failback策略
failsafe策略
failfast策略
available策略

现在我们来看下每个类之间的关系图：

failover策略

failover策略是dubbo默认的容错策略，即失败重试策略。通过上图的类关系图我们可以得知FailoverClusterInvoker的父类是AbstractClusterInvoker。现在我们来结合代码学习一下这种策略：

父类AbstractClusterInvoker的invoke方法其实没做较多处理，仅仅是获取了一下负载均衡策略。通过代码我们可以看到如果invoker列表中含有多个，就获取第一个invoker列表的负载策略，默认是random。现在我们看下子类实现的doInvoke方法：

        public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
    List<Invoker<T>> copyinvokers = invokers;
    /* 校验invokers列表是否为null，是的话则抛出异常 */
    this.checkInvokers(invokers, invocation);
    /* 获取重试次数，+1是因为第一次调用不算 */
    int len = this.getUrl().getMethodParameter(invocation.getMethodName(), "retries", 2) + 1;
    if (len <= 0) {
        len = 1;
    }


    RpcException le = null;
    List<Invoker<T>> invoked = new ArrayList(invokers.size());
    Set<String> providers = new HashSet(len);


    for(int i = 0; i < len; ++i) {
        /* 重试时，校验invoker列表会否发生变化，如果发生变化，则抛出异常 */
        if (i > 0) {
            this.checkWheatherDestoried();
            copyinvokers = this.list(invocation);
            this.checkInvokers(copyinvokers, invocation);
        }
        /* 通过负载均衡获取下一个服务提供者 */
        Invoker<T> invoker = this.select(loadbalance, invocation, copyinvokers, invoked);
        invoked.add(invoker);
        RpcContext.getContext().setInvokers(invoked);


        try {
            Result result = invoker.invoke(invocation);
            if (le != null && logger.isWarnEnabled()) {
                logger.warn("Although retry the method " + invocation.getMethodName() + " in the service " + this.getInterface().getName() + " was successful by the provider " + invoker.getUrl().getAddress() + ", but there have been failed providers " + providers + " (" + providers.size() + "/" + copyinvokers.size() + ") from the registry " + this.directory.getUrl().getAddress() + " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version " + Version.getVersion() + ". Last error is: " + le.getMessage(), le);
            }


            Result var12 = result;
            return var12;
        } catch (RpcException var18) {
            if (var18.isBiz()) {/* 如果是业务异常，直接抛出异常 */
                throw var18;
            }


            le = var18;
        } catch (Throwable var19) {
            le = new RpcException(var19.getMessage(), var19);
        } finally {
            providers.add(invoker.getUrl().getAddress());/* 记录调用失败信息 */
        }
    }


    throw new RpcException(le != null ? le.getCode() : 0, "Failed to invoke the method " + 
    invocation.getMethodName() + " in the service " + this.getInterface().getName() + ". Tried " + len + 
    " times of the providers " + providers + " (" + providers.size() + "/" + copyinvokers.size() + 
    ") from the registry " + this.directory.getUrl().getAddress() + " on the consumer " + 
    NetUtils.getLocalHost() + " using the dubbo version " + Version.getVersion() + ". Last error is: " + 
    (le != null ? le.getMessage() : ""), (Throwable)(le != null && le.getCause() != null ? le.getCause() : le));
}

现在来总结一下Failover策略，我们可以看到，首先会获取到重试次数，以这个为限制，如果每次调用失败，就用过负载均衡策略获取到下一个服务提供者，知道调用成功。或者方法调用受到重试次数的限制，抛出异常。如果中间有业务异常，则直接抛出异常，不再重试。

broadcast策略：

这种策略主要用在通知所有提供者更新缓存或日志等本地资源信息，现在我们来看下代码：

通过上图我们可以看到在校验完invokers列表后，会循环把所有的invoker调用一遍，覆盖掉以前的结果。如果其中有一个失败，则抛出异常。

Failback策略：

Failback策略是失败之后会自动回复，并在等待一段时间之后利用调度器进行重试。现在我们来看下代码：

我们可以看到仍然是校验完invokers列表后，直接调用select通过负载的方法获取invoker对象，如果invoker失败，则调用addFailed方法，加入调度器。现在来看下此方法的代码：

我们可以看到通过双检锁来保证创建的线程池的单例化，此线程池每5秒执行一次restryFailed方法，把实例放入map中。接下来看下restryFailed方法，这才是调用的核心：

此方法主要是从failied中获取到调用环境栈，并循环执行调用，如果调用成功则移除，失败则打印日志。

failsafe策略：

Failsafe策略如果在调用过程中出现异常，记录异常信息，不对异常做处理。通常可用于写入审计日志操作。下面我们来看下代码：

看代码逻辑是比较简单的，通过负载策略选出一个invoker对象进行调用，如果出现异常记录异常，然后空值。

failfast策略：

Failfast策略是只发生一次调用，如若捕获到异常，则直接抛出异常。可用于非幂等的操作，比如新增记录。现在来看下代码：

看代码逻辑是比较简单的，通过负载策略选出一个invoker对象进行调用，如果出现异常则直接抛出异常，终止操作。

available策略：

我们可以看到available策略是简单的调用第一个到达的服务。当都不可达时，则抛出异常。

现在我们来回想一下创建代理时的过程，当invokers其中有一个url的协议头等于registry时，就把容错策略设为available。所以再结合上图我们可以看到需要获取到一个可用的服务，如若为获取到就抛出异常。获取到就进行调用。

forking策略：

这种策略是并行调用多个服务器，只要一个成功就返回，主要应用在实时性较高的场景。但是如果并行个数较多，会消耗服务器资源。所以一般都是设置为fork=2。现在来看下代码：

public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
    this.checkInvokers(invokers, invocation);
    /* 获取并行获取服务器个数，默认是2 */
    int forks = this.getUrl().getParameter("forks", 2);
    /* 超时时间，默认是1s */
    int timeout = this.getUrl().getParameter("timeout", 1000);
    final Object selected;
    if (forks > 0 && forks < invokers.size()) {
        selected = new ArrayList();
        /* 通过负载均衡策略选出要并行调用的invoker，并放入selected列表 */
        for(int i = 0; i < forks; ++i) {
            /*在invoker列表(排除selected)后,如果没有选够,则存在重复循环问题.见select实现.*/
            Invoker<T> invoker = this.select(loadbalance, invocation, invokers, (List)selected);
            if (!((List)selected).contains(invoker)) {/* 防止重复添加 */
                ((List)selected).add(invoker);
            }
        }
    } else {
        selected = invokers;
    }


    RpcContext.getContext().setInvokers((List)selected);
    final AtomicInteger count = new AtomicInteger();
    final BlockingQueue<Object> ref = new LinkedBlockingQueue();
    Iterator i$ = ((List)selected).iterator();
    /* 遍历selected列表，通过线程池并发调用 */
    while(i$.hasNext()) {
        final Invoker<T> invoker = (Invoker)i$.next();
        this.executor.execute(new Runnable() {
            public void run() {
                try {
                    Result result = invoker.invoke(invocation);
                    ref.offer(result);
                } catch (Throwable var3) {
                    int value = count.incrementAndGet();
                    /* 所有的都异常了，才把异常加入到对了尾部
                     这就保证了，只要有一个成功，ref.poll()方法从队列头部就能取得到结果返回。 */
                    if (value >= ((List)selected).size()) {
                        ref.offer(var3);
                    }
                }


            }
        });
    }


    try {
        /* 从队列头部获取结果返回，如若是异常，就抛出 */
        Object ret = ref.poll((long)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);
        } else {
            return (Result)ret;
        }
    } catch (InterruptedException var11) {
        throw new RpcException("Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. Last error is: " + var11.getMessage(), var11);
    }
}

通过对Forking策略代码的学习，我们可以看到在获取到并行服务器个数后，通过线程池和阻塞队列来保证了并发安全性。在处理完成之后，异步的获取返回结果。如若是异常，则直接抛出。

如果仔细回想，我们可以观察到每个负载均衡策略都调用了select方法，接下来我们就对这个方法进行一个学习：

protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
    if (invokers != null && invokers.size() != 0) {
        /* 如果invocation为null的话，就返回空否则返回invocation的方法名 */
        String methodName = invocation == null ? "" : invocation.getMethodName();
        /* 是否启用sticky粘滞连接，让客户端总是连接一个提供者 */
        boolean sticky = ((Invoker)invokers.get(0)).getUrl().getMethodParameter(methodName, "sticky", false);
        /* 如果粘滞连接的可提供者列表null并且invokers不包含这个可提供者，则设为Null */
        if (this.stickyInvoker != null && !invokers.contains(this.stickyInvoker)) {
            this.stickyInvoker = null;
        }
        /* 如果允许粘滞连接，并且stickyInvoker不为Null，并且没有在已选列表中。并校验这个stic
         kyInvoker的可用性。如果为真，则直接返回stickyInvoker
         由于stickyInvoker不能包含在selected列表中，所以forking和failover策略没有办法采用粘滞连接 */
        if (sticky && this.stickyInvoker != null && (selected == null || !selected.contains(this.stickyInvoker)) 
            && this.availablecheck && this.stickyInvoker.isAvailable()) {
            return this.stickyInvoker;
        } else {
            /* 否则利用负载均衡策略选择一个invoker，如果需要粘滞连接，
             则把invoker赋值给stickyInvoker，并返回 */
            Invoker<T> invoker = this.doselect(loadbalance, invocation, invokers, selected);
            if (sticky) {
                this.stickyInvoker = invoker;
            }


            return invoker;
        }
    } else {
        return null;
    }
}

通过select方法，可以总结出如果允许粘滞连接，那么则粘滞连接的提供者不允许出现在已选列表中并是可用的。如果达到要求，则返回stickyInvoker，否则则通过doSelect选择出一个invoker。接下来看下doSelect方法：

private Invoker<T> doselect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
    if (invokers != null && invokers.size() != 0) {
        if (invokers.size() == 1) {/* 如果invoker列表长度为1，则直接获取第一个返回 */
            return (Invoker)invokers.get(0);
        /* 如果invoker列表长度为2并且selected列表长度>0，则判断selected列表的第一个和invoekr
         列表的第一个是否相等，为真返回invoker第二个，否则返回第一个 */    
        } else if (invokers.size() == 2 && selected != null && selected.size() > 0) {
            return selected.get(0) == invokers.get(0) ? (Invoker)invokers.get(1) : (Invoker)invokers.get(0);
        } else {
            /* 大于两个的话利用负载均衡选择一个 */
            Invoker<T> invoker = loadbalance.select(invokers, this.getUrl(), invocation);
            /* invoker包含在已选列表中或者invoker不可用，则重新选择 */
            if (selected != null && selected.contains(invoker) || !invoker.isAvailable() 
              && this.getUrl() != null && this.availablecheck) {
                try {
                    /* 重新选择 */
                    Invoker<T> rinvoker = this.reselect(loadbalance, invocation, invokers, selected, this.availablecheck);
                    if (rinvoker != null) {
                        invoker = rinvoker;
                    } else {
                        /* 如果重新选择失败，看下一次选的位置，如果不是最后，选+1位置 */
                        int index = invokers.indexOf(invoker);


                        try {
                            /* 避免碰撞 */
                            invoker = index < invokers.size() - 1 ? (Invoker)invokers.get(index + 1) : invoker;
                        } catch (Exception var9) {
                            logger.warn(var9.getMessage() + " may because invokers list dynamic change, ignore.", var9);
                        }
                    }
                } catch (Throwable var10) {
                    logger.error("clustor relselect fail reason is :" + var10.getMessage() + " if can not slove ,you can set cluster.availablecheck=false in url", var10);
                }
            }


            return invoker;
        }
    } else {
        return null;
    }
}

我们可以看到在doSelect方法在获取invoker对象时，主要是通过invokers列表的长度来进行选择的。如果长度为1，则直接获取第一个。长度等于两个，则轮询。如果长度>2，则利用负载均衡选择一个。当选择出来的invoker在已选列表中，并且不可用时，重新选择。否则直接返回invoker对象。重新选择失败的话获取到下一次选择的位置，如果不是最后一位，则获取+1位置的invoker，否则直接获取invoker。现在我们来看下重新选择reselect方法，这个方法的作用就是尽量不要从已选列表中选择invoker。

这里我们对dubbo框架中的其中集群容错策略进行了学习。dubbo默认的容错策略是失败重试。可以看到dubbo的每种策略所应用的场景是不尽相同的。

数据库

文章转载自Lord Lean Notes，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

dubbo-------初识dubbo集群容错机制

评论