Dubbo源码：集群容错

客官°小女子只卖身不卖艺 2022-10-16 02:54 239阅读 0赞

Dubbo源码：集群容错

    • 大致流程
      • FailoverClusterInvoker：失败切换
      • FailbackClusterInvoker：失败恢复
      • FailfastClusterInvoker：快速失败
      • FailsafeClusterInvoker：失败安全
      • ForkingClusterInvoker：并行调用多个服务提供者
      • BroadcastClusterInvoker：会逐个调用每个服务提供者

大致流程

消费者在生成代理对象后，调用RPC服务的方法，会执行代理对象的方法，因为创建代理对象的时候会传入一个InvocationHandler对象，这个对象是代理对象的一个属性，并且这个对象持有Invoker对象，Invoker对象是Dubbo领域模型中的核心模型，是实体域，包含这个RPC服务的所有信息。
代理对象对象中其实就是调用这个Invoker的invoker()方法，所以就会执行到AbstractClusterInvoker类的invoker()方法，这个方法是集群容错的父类，里面封装了集群容错会用到的一些公共方法，比如调用list()从RegistryDirectory中获取到获取Invoker列表，选择一个负载均衡器（LoadBalance），还有select()也就是负载均衡方法等等。

AbstractClusterInvoker还有一个doInvoke()的模板方法，具体实现调用逻辑和容错逻辑子类实现。下面就来看下它的几个主要实现类。

FailoverClusterInvoker：失败切换

是默认的容错机制，会先根据配置获取配置的重试次数，默认是3次。然后循环调用，如果不是第一次调用会调用list()重选列举服务目录。然后负载均衡选择Invoker，并且把选中后的Invoker记录下来，当失败重试的时候如果再次选到，会重新选择一次。最后调用Invoker的invoker()，成功则返回结果。如果调用失败会记录异常信息，然后重试。直到超出重试次数后，会抛出异常，告诉服务调用失败了。
通常用于读操作，但重试会带来更长延迟。 可通过retries=”2”来设置重试次数（不含第一次）。

  1. public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { 
  2.         List<Invoker<T>> copyinvokers = invokers;
  3.         checkInvokers(copyinvokers, invocation);
  5.         // 获取重试次数
  6.         int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
  7.         if (len <= 0) { 
  8.             len = 1;
  9.         }
  10.         // retry loop.
  11.         RpcException le = null; // last exception.
  12.         List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size()); // invoked invokers.
  13.         Set<String> providers = new HashSet<String>(len);
  14.         // 循环调用，失败重试
  15.         for (int i = 0; i < len; i++) { 
  17.             if (i > 0) { 
  18.                 checkWhetherDestroyed();
  19.                 // 在进行重试前重新列举 Invoker，这样做的好处是，如果某个服务挂了，
  20.                 // 通过调用 list 可得到最新可用的 Invoker 列表
  21.                 copyinvokers = list(invocation);
  22.                 // check again
  23.                 checkInvokers(copyinvokers, invocation);
  24.             }
  25.             // 通过负载均衡选择 Invoker
  26.             Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
  27.             // 选择后的记录下来，失败重试如果选到的在这里面会进行重选
  28.             invoked.add(invoker);
  29.             RpcContext.getContext().setInvokers((List) invoked);
  30.             try { 
  31.                 // 调用
  32.                 Result result = invoker.invoke(invocation);
  33.                 if (le != null && logger.isWarnEnabled()) { 
  34.                     logger.warn("....");
  35.                 }
  36.                 return result;
  37.             } catch (RpcException e) { 
  38.                 if (e.isBiz()) {  // biz exception.
  39.                     throw e;
  40.                 }
  41.                 le = e;
  42.             } catch (Throwable e) { 
  43.                 le = new RpcException(e.getMessage(), e);
  44.             } finally { 
  45.                 providers.add(invoker.getUrl().getAddress());
  46.             }
  47.         }
  48.         // 若重试失败，则抛出异常
  49.         throw new RpcException("....");
  50.     }

FailbackClusterInvoker：失败恢复

会在调用失败后，把任务放入到一个ConcurrentMap中，并启动一个延时任务去消费这个Map，最后返回一个空结果给服务消费者。适合执行消息通知等操作。

  1. protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { 
  2.         try { 
  3.             checkInvokers(invokers, invocation);
  4.             // 负载均衡算法，选择Invoker
  5.             Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
  6.             // 调用
  7.             return invoker.invoke(invocation);
  8.         } catch (Throwable e) { 
  9.             // 如果调用过程中发生异常，此时仅打印错误日志，不抛出异常
  10.             logger.error("...");
  11.             // 添加到失败列表中，定时重试
  12.             addFailed(invocation, this);
  13.             // 返回空结果
  14.             return new RpcResult(); // ignore
  15.         }
  16.     }
  17. private void addFailed(Invocation invocation, AbstractClusterInvoker<?> router) { 
  18.         if (retryFuture == null) { 
  19.             synchronized (this) { 
  20.                 if (retryFuture == null) { 
  21.                     retryFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() { 
  23.                         public void run() { 
  24.                             // collect retry statistics
  25.                             try { 
  26.                                 // 五秒重试
  27.                                 retryFailed();
  28.                             } catch (Throwable t) {  // Defensive fault tolerance
  29.                                 logger.error("Unexpected error occur at collect statistic", t);
  30.                             }
  31.                         }
  32.                     }, RETRY_FAILED_PERIOD, RETRY_FAILED_PERIOD, TimeUnit.MILLISECONDS);
  33.                 }
  34.             }
  35.         }
  36.         failed.put(invocation, router);
  37.     }
  39.     void retryFailed() { 
  40.         if (failed.size() == 0) { 
  41.             return;
  42.         }
  43.         // 重试失败列表
  44.         for (Map.Entry<Invocation, AbstractClusterInvoker<?>> entry : new HashMap<Invocation, AbstractClusterInvoker<?>>(failed).entrySet()) { 
  45.             Invocation invocation = entry.getKey();
  46.             Invoker<?> invoker = entry.getValue();
  47.             try { 
  48.                 invoker.invoke(invocation);
  49.                 failed.remove(invocation);
  50.             } catch (Throwable e) { 
  51.                 logger.error("....");
  52.             }
  53.         }
  54.     }

FailfastClusterInvoker：快速失败

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

  1. public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { 
  2.         checkInvokers(invokers, invocation);
  3.         Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
  4.         try { 
  5.             return invoker.invoke(invocation);
  6.         } catch (Throwable e) { 
  7.             if (e instanceof RpcException && ((RpcException) e).isBiz()) {  // biz exception.
  8.                 throw (RpcException) e;
  9.             }
  10.             throw new RpcException(e);
  11.         }
  12.     }

FailsafeClusterInvoker：失败安全

当调用过程中出现异常时，FailsafeClusterInvoker 仅会打印异常，而不会抛出异常。适用于写入审计日志等操作。

  1. public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { 
  2.         try { 
  3.             checkInvokers(invokers, invocation);
  4.             Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
  5.             return invoker.invoke(invocation);
  6.         } catch (Throwable e) { 
  7.             logger.error("Failsafe ignore exception: " + e.getMessage(), e);
  8.             return new RpcResult(); // ignore
  9.         }
  10.     }

ForkingClusterInvoker：并行调用多个服务提供者

会在运行时通过线程池创建多个线程，并发调用多个服务提供者，然后把结果放入BlockingQueue中，如果所有都调用失败，Queue中放的就是失败的异常信息。只要有一个服务提供者成功返回了，Queue中放的就是结果，最后从阻塞队列中取结果，如果取到的结果是异常就抛出，是结果就正常返回。
主要应用场景是在一些对实时性要求比较高读操作。

  1. public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { 
  2.         checkInvokers(invokers, invocation);
  3.         final List<Invoker<T>> selected;
  4.         final int forks = getUrl().getParameter(Constants.FORKS_KEY, Constants.DEFAULT_FORKS);
  5.         final int timeout = getUrl().getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
  6.         if (forks <= 0 || forks >= invokers.size()) { 
  7.             selected = invokers;
  8.         } else { 
  9.             selected = new ArrayList<Invoker<T>>();
  10.             for (int i = 0; i < forks; i++) { 
  11.                 // TODO. Add some comment here, refer chinese version for more details.
  12.                 Invoker<T> invoker = select(loadbalance, invocation, invokers, selected);
  13.                 if (!selected.contains(invoker)) { //Avoid add the same invoker several times.
  14.                     selected.add(invoker);
  15.                 }
  16.             }
  17.         }
  18.         RpcContext.getContext().setInvokers((List) selected);
  19.         // 记录失败次数
  20.         final AtomicInteger count = new AtomicInteger();
  21.         // 记录成功或者异常信息
  22.         final BlockingQueue<Object> ref = new LinkedBlockingQueue<Object>();
  23.         for (final Invoker<T> invoker : selected) { 
  24.             executor.execute(new Runnable() { 
  25.                 public void run() { 
  26.                     try { 
  27.                         // 执行
  28.                         Result result = invoker.invoke(invocation);
  29.                         ref.offer(result);
  30.                     } catch (Throwable e) { 
  31.                         int value = count.incrementAndGet();
  32.                         if (value >= selected.size()) { 
  33.                             ref.offer(e);
  34.                         }
  35.                     }
  36.                 }
  37.             });
  38.         }
  39.         try { 
  40.             // 阻塞获取结果，如果是异常类型，就抛出异常
  41.             Object ret = ref.poll(timeout, TimeUnit.MILLISECONDS);
  42.             if (ret instanceof Throwable) { 
  43.                 Throwable e = (Throwable) ret;
  44.                 throw new RpcException(e);
  45.             }
  46.             return (Result) ret;
  47.         } catch (InterruptedException e) { 
  48.             throw new RpcException("");
  49.         }
  50.     }

BroadcastClusterInvoker：会逐个调用每个服务提供者

如果其中一台报错，在循环调用结束后，BroadcastClusterInvoker 会抛出异常。该类通常用于通知所有提供者更新缓存或日志等本地资源信息

  1. public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { 
  2.         checkInvokers(invokers, invocation);
  3.         RpcContext.getContext().setInvokers((List) invokers);
  4.         RpcException exception = null;
  5.         Result result = null;
  6.         for (Invoker<T> invoker : invokers) { 
  7.             try { 
  8.                 result = invoker.invoke(invocation);
  9.             } catch (RpcException e) { 
  10.                 // 记录异常
  11.                 exception = e;
  12.                 logger.warn(e.getMessage(), e);
  13.             } catch (Throwable e) { 
  14.                 exception = new RpcException(e.getMessage(), e);
  15.                 logger.warn(e.getMessage(), e);
  16.             }
  17.         }
  18.         if (exception != null) { 
  19.             throw exception;
  20.         }
  21.         return result;
  22.     }

发表评论

表情：
评论列表 （有 0 条评论，239人围观）

还没有评论，来说两句吧...

相关阅读