物业公司网站建设,有了域名如何建设网站,品牌策划的七个步骤,长春专业网站推广3.2.2 FailbackClusterInvoker
FailbackClusterInvoker 会在调用失败后#xff0c;返回一个空结果给服务提供者。并通过定时任务对失败的调用进行重传#xff0c;适合执行消息通知等操作。下面来看一下它的实现逻辑。
public class FailbackClusterInvokerT extend…3.2.2 FailbackClusterInvoker
FailbackClusterInvoker 会在调用失败后返回一个空结果给服务提供者。并通过定时任务对失败的调用进行重传适合执行消息通知等操作。下面来看一下它的实现逻辑。
public class FailbackClusterInvokerT extends AbstractClusterInvokerT {private static final long RETRY_FAILED_PERIOD 5 * 1000;private final ScheduledExecutorService scheduledExecutorService Executors.newScheduledThreadPool(2,new NamedInternalThreadFactory(failback-cluster-timer, true));private final ConcurrentMapInvocation, AbstractClusterInvoker? failed new ConcurrentHashMapInvocation, AbstractClusterInvoker?();private volatile ScheduledFuture? retryFuture;Overrideprotected Result doInvoke(Invocation invocation, ListInvokerT invokers, LoadBalance loadbalance) throws RpcException {try {checkInvokers(invokers, invocation);// 选择 InvokerInvokerT invoker select(loadbalance, invocation, invokers, null);// 进行调用return invoker.invoke(invocation);} catch (Throwable e) {// 如果调用过程中发生异常此时仅打印错误日志不抛出异常logger.error(Failback to invoke method ...);// 记录调用信息addFailed(invocation, this);// 返回一个空结果给服务消费者return new RpcResult();}}private void addFailed(Invocation invocation, AbstractClusterInvoker? router) {if (retryFuture null) {synchronized (this) {if (retryFuture null) {// 创建定时任务每隔5秒执行一次retryFuture scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {Overridepublic void run() {try {// 对失败的调用进行重试retryFailed();} catch (Throwable t) {// 如果发生异常仅打印异常日志不抛出logger.error(Unexpected error occur at collect statistic, t);}}}, RETRY_FAILED_PERIOD, RETRY_FAILED_PERIOD, TimeUnit.MILLISECONDS);}}}// 添加 invocation 和 invoker 到 failed 中// 这里的把 invoker 命名为 router很奇怪明显名不副实failed.put(invocation, router);}void retryFailed() {if (failed.size() 0) {return;}// 遍历 failed对失败的调用进行重试for (Map.EntryInvocation, AbstractClusterInvoker? entry : new HashMapInvocation, AbstractClusterInvoker?(failed).entrySet()) {Invocation invocation entry.getKey();Invoker? invoker entry.getValue();try {// 再次进行调用invoker.invoke(invocation);// 调用成功则从 failed 中移除 invokerfailed.remove(invocation);} catch (Throwable e) {// 仅打印异常不抛出logger.error(Failed retry to invoke method ...);}}}
}
这个类主要由3个方法组成首先是 doInvoker该方法负责初次的远程调用。若远程调用失败则通过 addFailed 方法将调用信息存入到 failed 中等待定时重试。addFailed 在开始阶段会根据 retryFuture 为空与否来决定是否开启定时任务。retryFailed 方法则是包含了失败重试的逻辑该方法会对 failed 进行遍历然后依次对 Invoker 进行调用。调用成功则将 Invoker 从 failed 中移除调用失败则忽略失败原因。
以上就是 FailbackClusterInvoker 的执行逻辑不是很复杂继续往下看。
3.2.3 FailfastClusterInvoker
FailfastClusterInvoker 只会进行一次调用失败后立即抛出异常。适用于幂等操作比如新增记录。楼主日常开发中碰到过一次程序连续插入三条同样的记录问题原因是新增记录过程中包含了一些耗时操作导致接口超时。而我当时使用的是 Dubbo 默认的 Cluster Invoker即 FailoverClusterInvoker。其会在调用失败后进行重试所以导致插入服务提供者插入了3条同样的数据。如果当时考虑使用 FailfastClusterInvoker就不会出现这种问题了。当然此时接口仍然会超时所以更合理的做法是使用 Dubbo 异步特性。或者优化服务逻辑避免超时。
其他的不多说了下面直接看源码吧。
public class FailfastClusterInvokerT extends AbstractClusterInvokerT {Overridepublic Result doInvoke(Invocation invocation, ListInvokerT invokers, LoadBalance loadbalance) throws RpcException {checkInvokers(invokers, invocation);// 选择 InvokerInvokerT invoker select(loadbalance, invocation, invokers, null);try {// 调用 Invokerreturn invoker.invoke(invocation);} catch (Throwable e) {if (e instanceof RpcException ((RpcException) e).isBiz()) {// 抛出异常throw (RpcException) e;}// 抛出异常throw new RpcException(..., Failfast invoke providers ...);}}
}
上面代码比较简单了首先是通过 select 方法选择 Invoker然后进行远程调用。如果调用失败则立即抛出异常。FailfastClusterInvoker 就先分析到这下面分析 FailsafeClusterInvoker。
3.2.4 FailsafeClusterInvoker
FailsafeClusterInvoker 是一种失败安全的 Cluster Invoker。所谓的失败安全是指当调用过程中出现异常时FailsafeClusterInvoker 仅会打印异常而不会抛出异常。Dubbo 官方给出的应用场景是写入审计日志等操作这个场景我在日常开发中没遇到过没发言权就不多说了。下面直接分析源码。
public class FailsafeClusterInvokerT extends AbstractClusterInvokerT {Overridepublic Result doInvoke(Invocation invocation, ListInvokerT invokers, LoadBalance loadbalance) throws RpcException {try {checkInvokers(invokers, invocation);// 选择 InvokerInvokerT invoker select(loadbalance, invocation, invokers, null);// 进行远程调用return invoker.invoke(invocation);} catch (Throwable e) {// 打印错误日志但不抛出logger.error(Failsafe ignore exception: e.getMessage(), e);// 返回空结果忽略错误return new RpcResult();}}
}
FailsafeClusterInvoker 的逻辑和 FailfastClusterInvoker 的逻辑一样简单因此就不多说了。继续下面分析。
3.2.5 ForkingClusterInvoker
ForkingClusterInvoker 会在运行时通过线程池创建多个线程并发调用多个服务提供者。只要有一个服务提供者成功返回了结果doInvoke 方法就会立即结束运行。ForkingClusterInvoker 的应用场景是在一些对实时性要求比较高读操作注意是读操作并行写操作可能不安全下使用但这将会耗费更多的服务资源。下面来看该类的实现。
public class ForkingClusterInvokerT extends AbstractClusterInvokerT {private final ExecutorService executor Executors.newCachedThreadPool(new NamedInternalThreadFactory(forking-cluster-timer, true));Overridepublic Result doInvoke(final Invocation invocation, ListInvokerT invokers, LoadBalance loadbalance) throws RpcException {try {checkInvokers(invokers, invocation);final ListInvokerT selected;// 获取 forks 配置final int forks getUrl().getParameter(Constants.FORKS_KEY, Constants.DEFAULT_FORKS);// 获取超时配置final int timeout getUrl().getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);// 如果 forks 配置不合理则直接将 invokers 赋值给 selectedif (forks 0 || forks invokers.size()) {selected invokers;} else {selected new ArrayListInvokerT();// 循环选出 forks 个 Invoker并添加到 selected 中for (int i 0; i forks; i) {// 选择 InvokerInvokerT invoker select(loadbalance, invocation, invokers, selected);if (!selected.contains(invoker)) {selected.add(invoker);}}}// ----------------------✨ 分割线1 ✨---------------------- //RpcContext.getContext().setInvokers((List) selected);final AtomicInteger count new AtomicInteger();final BlockingQueueObject ref new LinkedBlockingQueueObject();// 遍历 selected 列表for (final InvokerT invoker : selected) {// 为每个 Invoker 创建一个执行线程executor.execute(new Runnable() {Overridepublic void run() {try {// 进行远程调用Result result invoker.invoke(invocation);// 将结果存到阻塞队列中ref.offer(result);} catch (Throwable e) {int value count.incrementAndGet();// 仅在 value 大于等于 selected.size() 时才将异常对象// 放入阻塞队列中请大家思考一下为什么要这样做。if (value selected.size()) {// 将异常对象存入到阻塞队列中ref.offer(e);}}}});}// ----------------------✨ 分割线2 ✨---------------------- //try {// 从阻塞队列中取出远程调用结果Object ret ref.poll(timeout, TimeUnit.MILLISECONDS);// 如果结果类型为 Throwable则抛出异常if (ret instanceof Throwable) {Throwable e (Throwable) ret;throw new RpcException(..., Failed to forking invoke provider ...);}// 返回结果return (Result) ret;} catch (InterruptedException e) {throw new RpcException(Failed to forking invoke provider ...);}} finally {RpcContext.getContext().clearAttachments();}}
}
ForkingClusterInvoker 的 doInvoker 方法比较长这里我通过两个分割线将整个方法划分为三个逻辑块。从方法开始到分割线1之间的代码主要是用于选出 forks 个 Invoker为接下来的并发调用提供输入。分割线1和分割线2之间的逻辑主要是通过线程池并发调用多个 Invoker并将结果存储在阻塞队列中。分割线2到方法结尾之间的逻辑主要用于从阻塞队列中获取返回结果并对返回结果类型进行判断。如果为异常类型则直接抛出否则返回。
以上就是ForkingClusterInvoker 的 doInvoker 方法大致过程。我在分割线1和分割线2之间的代码上留了一个问题问题是这样的为什么要在 value selected.size() 的情况下才将异常对象添加到阻塞队列中这里来解答一下。原因是这样的在并行调用多个服务提供者的情况下哪怕只有一个服务提供者成功返回结果而其他全部失败。此时 ForkingClusterInvoker 仍应该返回成功的结果而非抛出异常。在 value selected.size() 时将异常对象放入阻塞队列中可以保证异常对象不会出现在正常结果的前面这样可从阻塞队列中优先取出正常的结果。
好了关于 ForkingClusterInvoker 就先分析到这接下来分析最后一个 Cluster Invoker。
3.2.6 BroadcastClusterInvoker
本章的最后我们再来看一下 BroadcastClusterInvoker。BroadcastClusterInvoker 会逐个调用每个服务提供者如果其中一台报错在循环调用结束后BroadcastClusterInvoker 会抛出异常。看官方文档上的说明该类通常用于通知所有提供者更新缓存或日志等本地资源信息。这个使用场景笔者也没遇到过没法详细说明了所以下面还是直接分析源码吧。
public class BroadcastClusterInvokerT extends AbstractClusterInvokerT {Overridepublic Result doInvoke(final Invocation invocation, ListInvokerT invokers, LoadBalance loadbalance) throws RpcException {checkInvokers(invokers, invocation);RpcContext.getContext().setInvokers((List) invokers);RpcException exception null;Result result null;// 遍历 Invoker 列表逐个调用for (InvokerT 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;}
}
以上就是 BroadcastClusterInvoker 的代码比较简单就不多说了。
4.总结
