Dubbo源码学习--MergeableCluster集群容错（九）

太过爱你忘了你带给我的痛 2022-05-28 01:26 231阅读 0赞

MergeableCluster聚合集群，将集群中的调用结果聚合起来返回结果。比如菜单服务，接口一样，但有多种实现，用group区分，现在消费方需从每种group中调用一次返回结果，合并结果返回，这样就可以实现聚合菜单项。比较鸡肋，小点的项目应该都用不到

配置如：(搜索所有分组)

  1. 按组合并返回结果，比如菜单服务，接口一样，但有多种实现，用group区分，现在消费方需从每种group中调用一次返回结果，合并结果返回，这样就可以实现聚合菜单项。比较鸡肋，小点的项目应该都用不到
  2. 配置如：(搜索所有分组)
  3. <dubbo:reference interface="com.xxx.MenuService" group="*" merger="true" />
  4. (合并指定分组)
  5. <dubbo:reference interface="com.xxx.MenuService" group="aaa,bbb" merger="true" />
  7. (指定方法合并结果，其它未指定的方法，将只调用一个Group)
  8. <dubbo:reference interface="com.xxx.MenuService" group="*">
  9.     <dubbo:method name="getMenuItems" merger="true" />
  10. </dubbo:service>
  12. (某个方法不合并结果，其它都合并结果)
  13. <dubbo:reference interface="com.xxx.MenuService" group="*" merger="true">
  14. <dubbo:method name="getMenuItems" merger="false" />
  15. </dubbo:service>
  17. (指定合并策略，缺省根据返回值类型自动匹配，如果同一类型有两个合并器时，需指定合并器的名称)
  18. <dubbo:reference interface="com.xxx.MenuService" group="*">
  19.     <dubbo:method name="getMenuItems" merger="mymerge" />
  20. </dubbo:service>
  22. (指定合并方法，将调用返回结果的指定方法进行合并，合并方法的参数类型必须是返回结果类型本身)
  23. <dubbo:reference interface="com.xxx.MenuService" group="*">
  24.     <dubbo:method name="getMenuItems" merger=".addAll" />
  25. </dubbo:service>

MergeableCluster的实现如下：

  1. public class MergeableCluster implements Cluster {
  3.     public static final String NAME = "mergeable";
  5.     public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
  6.         return new MergeableClusterInvoker<T>(directory);
  7.     }
  9. }

结果调用及聚合操作是在MergeableClusterInvoker中完成实现的。

  1. @SuppressWarnings("unchecked")
  2. public class MergeableClusterInvoker<T> implements Invoker<T> {
  4.     private static final Logger log = LoggerFactory.getLogger(MergeableClusterInvoker.class);
  5.     private final Directory<T> directory;
  6.     private ExecutorService executor = Executors.newCachedThreadPool(new NamedThreadFactory("mergeable-cluster-executor", true));
  8.     public MergeableClusterInvoker(Directory<T> directory) {
  9.         this.directory = directory;
  10.     }
  12.     @SuppressWarnings("rawtypes")
  13.     public Result invoke(final Invocation invocation) throws RpcException {
  14.         List<Invoker<T>> invokers = directory.list(invocation);
  16.         String merger = getUrl().getMethodParameter(invocation.getMethodName(), Constants.MERGER_KEY);
  17.         //如果不存在分组，则只调用一个即可
  18.         if (ConfigUtils.isEmpty(merger)) { // If a method doesn't have a merger, only invoke one Group
  19.             for (final Invoker<T> invoker : invokers) {
  20.                 if (invoker.isAvailable()) {
  21.                     return invoker.invoke(invocation);
  22.                 }
  23.             }
  24.             return invokers.iterator().next().invoke(invocation);
  25.         }
  27.         Class<?> returnType;
  28.         try {
  29.             returnType = getInterface().getMethod(
  30.                     invocation.getMethodName(), invocation.getParameterTypes()).getReturnType();
  31.         } catch (NoSuchMethodException e) {
  32.             returnType = null;
  33.         }
  34.         //如果存在分组则调用所有的分组，将结果添加到results中
  35.         Map<String, Future<Result>> results = new HashMap<String, Future<Result>>();
  36.         for (final Invoker<T> invoker : invokers) {
  37.             Future<Result> future = executor.submit(new Callable<Result>() {
  38.                 public Result call() throws Exception {
  39.                     return invoker.invoke(new RpcInvocation(invocation, invoker));
  40.                 }
  41.             });
  42.             results.put(invoker.getUrl().getServiceKey(), future);
  43.         }
  45.         Object result = null;
  47.         List<Result> resultList = new ArrayList<Result>(results.size());
  49.         //合并结果值
  50.         int timeout = getUrl().getMethodParameter(invocation.getMethodName(), Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
  51.         for (Map.Entry<String, Future<Result>> entry : results.entrySet()) {
  52.             Future<Result> future = entry.getValue();
  53.             try {
  54.                 Result r = future.get(timeout, TimeUnit.MILLISECONDS);
  55.                 if (r.hasException()) {
  56.                     log.error(new StringBuilder(32).append("Invoke ")
  57.                                     .append(getGroupDescFromServiceKey(entry.getKey()))
  58.                                     .append(" failed: ")
  59.                                     .append(r.getException().getMessage()).toString(),
  60.                             r.getException());
  61.                 } else {
  62.                     resultList.add(r);
  63.                 }
  64.             } catch (Exception e) {
  65.                 throw new RpcException(new StringBuilder(32)
  66.                         .append("Failed to invoke service ")
  67.                         .append(entry.getKey())
  68.                         .append(": ")
  69.                         .append(e.getMessage()).toString(),
  70.                         e);
  71.             }
  72.         }
  74.         if (resultList.isEmpty()) {
  75.             return new RpcResult((Object) null);
  76.         } else if (resultList.size() == 1) {
  77.             return resultList.iterator().next();
  78.         }
  80.         if (returnType == void.class) {
  81.             return new RpcResult((Object) null);
  82.         }
  83.         //分组以.开头则只返回一个结果值
  84.         if (merger.startsWith(".")) {
  85.             merger = merger.substring(1);
  86.             Method method;
  87.             try {
  88.                 method = returnType.getMethod(merger, returnType);
  89.             } catch (NoSuchMethodException e) {
  90.                 throw new RpcException(new StringBuilder(32)
  91.                         .append("Can not merge result because missing method [ ")
  92.                         .append(merger)
  93.                         .append(" ] in class [ ")
  94.                         .append(returnType.getClass().getName())
  95.                         .append(" ]")
  96.                         .toString());
  97.             }
  98.             if (!Modifier.isPublic(method.getModifiers())) {
  99.                 method.setAccessible(true);
  100.             }
  101.             result = resultList.remove(0).getValue();
  102.             try {
  103.                 if (method.getReturnType() != void.class
  104.                         && method.getReturnType().isAssignableFrom(result.getClass())) {
  105.                     for (Result r : resultList) {
  106.                         result = method.invoke(result, r.getValue());
  107.                     }
  108.                 } else {
  109.                     for (Result r : resultList) {
  110.                         method.invoke(result, r.getValue());
  111.                     }
  112.                 }
  113.             } catch (Exception e) {
  114.                 throw new RpcException(
  115.                         new StringBuilder(32)
  116.                                 .append("Can not merge result: ")
  117.                                 .append(e.getMessage()).toString(),
  118.                         e);
  119.             }
  120.         } else {
  121.             //返回所有的结果值
  122.             Merger resultMerger;
  123.             if (ConfigUtils.isDefault(merger)) {
  124.                 resultMerger = MergerFactory.getMerger(returnType);
  125.             } else {
  126.                 resultMerger = ExtensionLoader.getExtensionLoader(Merger.class).getExtension(merger);
  127.             }
  128.             if (resultMerger != null) {
  129.                 List<Object> rets = new ArrayList<Object>(resultList.size());
  130.                 for (Result r : resultList) {
  131.                     rets.add(r.getValue());
  132.                 }
  133.                 result = resultMerger.merge(
  134.                         rets.toArray((Object[]) Array.newInstance(returnType, 0)));
  135.             } else {
  136.                 throw new RpcException("There is no merger to merge result.");
  137.             }
  138.         }
  139.         return new RpcResult(result);
  140.     }
  142.     public Class<T> getInterface() {
  143.         return directory.getInterface();
  144.     }
  146.     public URL getUrl() {
  147.         return directory.getUrl();
  148.     }
  150.     public boolean isAvailable() {
  151.         return directory.isAvailable();
  152.     }
  154.     public void destroy() {
  155.         directory.destroy();
  156.     }
  158.     private String getGroupDescFromServiceKey(String key) {
  159.         int index = key.indexOf("/");
  160.         if (index > 0) {
  161.             return new StringBuilder(32).append("group [ ")
  162.                     .append(key.substring(0, index)).append(" ]").toString();
  163.         }
  164.         return key;
  165.     }
  166. }

发表评论

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

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

相关阅读