ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > IT职业 > IT生活 > dubbo源码-集群容错之MergeableCluster

dubbo源码-集群容错之MergeableCluster

IT生活 作者：QCHuangChao 时间：2021-09-09 09:19:26 0 删除 编辑

简介

在dubbo官方的用户手册中，提到了使用MergeableCluster的场景--分组聚合：

按组合并返回结果 ，比如菜单服务，接口一样，但有多种实现，用group区分，现在消费者需从每种group中调用一次返回结果，合并结果返回，这样就可以实现聚合菜单项。

功能示意图如下：


图片描述

merger示意图

用法

定义菜单接口方式：

/**
 * @author wangzhenfei9
 * @version 1.0.0
 * @since 2018年05月25日
 */public interface MenuService {    List getMenuList(String traceId);

}

这个接口有两个实现类：HotMenuServiceImpl和ColdMenuServiceImpl；前者返回结果为：[{"id":1,"name":"青椒炒肉"},{"id":2,"name":"剁椒鱼头"},{"id":3,"name":"口味虾"}]，后者返回结果为：[{"id":101,"name":"凉拌黄瓜"},{"id":102,"name":"凉拌木耳"}]；

Provider暴露服务--一个服务属于group-hot，一个服务属于group-cold：

笔者测试时启动了两个Provider，所以总计有四个服务，dubbo-monitor监控显示如下：


图片描述

dubbo-monitor服务列表

Consumer调用服务：

几个重要的配置说明：

  • merger: merger="list"指定merge方式，可以自定义，也可以指定com.alibaba.dubbo.rpc.cluster.Merger文件中申明的方式；

  • group: group="*"表示调用接口com.alibaba.dubbo.demo.MenuService所有的分组服务，由于只有group-hot和group-cold两个分组，这里也可以配置为group="group-hot,group-cold"；

  • cluster: cluster="mergeable"即指定集群容错模式为MergeableCluster模式，也就是本文分析的模式；

  • timeout: timeout="1800000"即超时时间，之所以设置这么大，是为了debug源码过程中不会发生超时，此配置不适用于生产环境；

com.alibaba.dubbo.rpc.cluster.Merger文件内容如下：

list=com.alibaba.dubbo.rpc.cluster.merger.ListMergerset=com.alibaba.dubbo.rpc.cluster.merger.SetMergermap=com.alibaba.dubbo.rpc.cluster.merger.MapMerger
byte=com.alibaba.dubbo.rpc.cluster.merger.ByteArrayMergerchar=com.alibaba.dubbo.rpc.cluster.merger.CharArrayMergershort=com.alibaba.dubbo.rpc.cluster.merger.ShortArrayMergerint=com.alibaba.dubbo.rpc.cluster.merger.IntArrayMergerlong=com.alibaba.dubbo.rpc.cluster.merger.LongArrayMergerfloat=com.alibaba.dubbo.rpc.cluster.merger.FloatArrayMergerdouble=com.alibaba.dubbo.rpc.cluster.merger.DoubleArrayMerger
boolean=com.alibaba.dubbo.rpc.cluster.merger.BooleanArrayMerger

这里需要指出的一点，dubbo官方在2017年11月份对这个文件有过修改，修改记录请戳：合并结果问题，应该是笔误，建议修复，修改内容如下图所示，所以老版本和新版本的merger="list"效果不一样：

图片描述

Merger bug fix


  • 运行结果
    [{"id":101,"name":"凉拌黄瓜"},{"id":102,"name":"凉拌木耳"},{"id":1,"name":"青椒炒肉"},{"id":2,"name":"剁椒鱼头"},{"id":3,"name":"口味虾"}]，从运行结果可以看出，合并了HotMenuServiceImpl和ColdMenuServiceImpl两个不同group服务的结果；

源码分析

核心源码在MergeableClusterInvoker.java中，源码如下所示：

@Override@SuppressWarnings("rawtypes")public Result invoke(final Invocation invocation) throws RpcException {    // 拿到可用的Invoker集合
    List> invokers = directory.list(invocation);    // 得到配置的merger参数值
    String merger = getUrl().getMethodParameter( invocation.getMethodName(), Constants.MERGER_KEY );    // 如果方法不需要Merge，退化为只调一个group即可--选择第一个有效的Invoker调用并返回结果
    if ( ConfigUtils.isEmpty(merger) ) {        for(final Invoker invoker : invokers ) {            if (invoker.isAvailable()) {                return invoker.invoke(invocation);
            }
        }        // 如果没有任意Invoker满足isAvailable(), 那么尝试调用第一个Invoker(多尝试一下, 多一次机会)
        return invokers.iterator().next().invoke(invocation);
    }    // 得到方法的返回类型
    Class> returnType;    try {
        returnType = getInterface().getMethod(
                invocation.getMethodName(), invocation.getParameterTypes() ).getReturnType();
    } catch ( NoSuchMethodException e ) {
        returnType = null;
    }    // 由于我们调用的服务, 有两个不同的group, 且没有申明version, 所以这个map的key有两个值
    Map> results = new HashMap>();    for( final Invoker invoker : invokers ) {        // 线程池方法异步调用
        Future future = executor.submit( new Callable() {            public Result call() throws Exception {                return invoker.invoke(new RpcInvocation(invocation, invoker));
            }
        } );        // serviceKey非常重要--serviceKey的值为: groupName/serviceInterface:version,
        // 如果version没有申明, serviceKey的值为: groupName/serviceInterface
        // 如果group没有申明, serviceKey的值为: serviceInterface:version
        // 如果version和group都没有申明, serviceKey的值为: serviceInterface
        results.put( invoker.getUrl().getServiceKey(), future );
    }

    Object result = null;    // 保存异步执行结果集合
    List resultList = new ArrayList( results.size() );    
    int timeout = getUrl().getMethodParameter( invocation.getMethodName(), Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT );    for ( Map.Entry> entry : results.entrySet() ) {
        Future future = entry.getValue();        try {
            Result r = future.get(timeout, TimeUnit.MILLISECONDS);            // 如果异步执行有异常(包括超时), 那么输出error级别的日志, 不影响最终的结果(只是部分数据缺失)
            if (r.hasException()) {
                log.error(new StringBuilder(32).append("Invoke ")
                              .append(getGroupDescFromServiceKey(entry.getKey()))
                              .append(" failed: ")
                              .append(r.getException().getMessage()).toString(),
                          r.getException());
            } else {
                resultList.add(r);
            }
        } catch ( Exception e ) {            throw new RpcException( new StringBuilder( 32 )
                                            .append( "Failed to invoke service " )
                                            .append( entry.getKey() )
                                            .append( ": " )
                                            .append( e.getMessage() ).toString(),
                                    e );
        }
    }    
    if (resultList.size() == 0) {        // 如果没有结果, 那么new一个result为null的RpcResult返回即可
        return new RpcResult((Object)null);
    } else if (resultList.size() == 1) {        // 如果只有一个结果, 那么直接返回即可
        return resultList.iterator().next();
    }    // 如果返回类型为void, 那么new一个result为null的RpcResult返回即可
    if (returnType == void.class) {        return new RpcResult((Object)null);
    }    // 如果merger的值是以.开头, 例如merger=".addAll", 这段逻辑就是调用结果类型的原生方法, 例如服务的返回结果是List，即list类型，那么merger=".addAll"就是调用List集合的.addAll()。
    if ( merger.startsWith(".") ) {
        merger = merger.substring(1);
        Method method;        try {            // 首先得到调用方法，如果方法不存在，则抛出异常
            method = returnType.getMethod( merger, returnType );
        } catch ( NoSuchMethodException e ) {            throw new RpcException(... ...);
        }        if ( method != null ) {            if ( !Modifier.isPublic( method.getModifiers() ) ) {
                method.setAccessible( true );
            }            // 先取得第一个结果
            result = resultList.remove( 0 ).getValue();            try {                // 如果merger=".addAll"指定的方法返回类型不为void，且和dubbo服务接口方法返回类型是相同类型，以测试代码为例，.addAll()返回类型是boolean，而dubbo服务接口方法返回类型是List，所以这里的if条件分支为false
                if ( method.getReturnType() != void.class
                        && method.getReturnType().isAssignableFrom( result.getClass() ) ) {                    // 遍历剩余的结果集
                    for ( Result r : resultList ) {                        // 根据配置的merger值，例如merger=".addAll"，依次对剩余结果集调用addAll()方法
                        result = method.invoke( result, r.getValue() );
                    }
                } else {                    // 遍历剩余的结果集
                    for ( Result r : resultList ) {                        // 根据配置的merger值，例如merger=".addAll"，依次对剩余结果集调用addAll()方法
                        method.invoke( result, r.getValue() );
                    }
                }
            } catch ( Exception e ) {                throw new RpcException( 
                        new StringBuilder( 32 )
                                .append( "Can not merge result: " )
                                .append( e.getMessage() ).toString(), 
                        e );
            }
        } else {            throw new RpcException(... ...);
        }
    } else {        // 如果merger申明为不以.开头, 例如merger="list"
        Merger resultMerger;        // true和default都是默认值(大小写不敏感)
        if (ConfigUtils.isDefault(merger)) {
            resultMerger = MergerFactory.getMerger(returnType);
        } else {            // 如果merger配置的是com.alibaba.dubbo.rpc.cluster.Merger文件中申明的值, 例如merger="list", 或者merger="map"
            resultMerger = ExtensionLoader.getExtensionLoader(Merger.class).getExtension(merger);
        }        if (resultMerger != null) {
            List rets = new ArrayList(resultList.size());            for(Result r : resultList) {
                rets.add(r.getValue());
            }            // 根据不用的merger实现, 合并结果
            result = resultMerger.merge(
                    rets.toArray((Object[])Array.newInstance(returnType, 0)));
        } else {            // 如果申明一些未知的merger, 那么抛出异常
            throw new RpcException( "There is no merger to merge result." );
        }
    }    return new RpcResult( result );
}

这一段代码还是很有借鉴意义的，比如支付宝获取支付方式（支付方式有多种，例如余额，红包，优惠券等），假设每种支付方式需要通过实时调用远程服务获取可用性，就可以模拟这种方式进行调用，美滋滋_

注意

在条件分支if ( merger.startsWith(".") ) {}中，有一段逻辑：method = returnType.getMethod( merger, returnType );，即从dubbo服务接口方法返回类型即java.util.List中查找merger配置的方法，例如.addAll，我们先看一下debug过程各变量的值：

图片描述

debug过程中变量的值


dubbo源码中method = returnType.getMethod( merger, returnType );调用Method method = getMethod0(name, parameterTypes, true);，再调用Method res =  privateGetMethodRecursive(name, parameterTypes, includeStaticMethods, interfaceCandidates);，最后调用searchMethods(privateGetDeclaredMethods(true), name, parameterTypes))，得到最后方法匹配的核心逻辑如下：

private static Method searchMethods(Method[] methods, String name, Class>[] parameterTypes){
    Method res = null;
    String internedName = name.intern();    for (int i = 0; i 
从searchMethods()源码可知，方法匹配需要满足几个条件：
    

  1. 方法名一样，即m.getName() == internedName。配置的是merger=".addAll"，而List中也有addAll方法，这个条件符合；
    2. 

  2. 寻找的方法参数类型和dubbo服务接口方法的返回类型完全一致（不能是继承完全），即arrayContentsEq(parameterTypes, m.getParameterTypes())。List中.addAll()方法参数类型是Collection（boolean addAll(Collection extends E> c);），而dubbo服务接口方法的返回类型是List类型，虽然List继承自Collection，但是并不等于，即arrayContentsEq()返回的还是false；
    3. 

由上面的分析可知，如果要merger=".addAll"能够正常工作，那么只需要将dubbo服务的返回类型改成Collection即可，例如：
Collection getMenuList(String traceId);
自定义merger实现
如果com.alibaba.dubbo.rpc.cluster.Merger文件集中方法无法满足需求，需要自定义实现，那么还是和dubbo其他扩展实现一样，依赖SPI。只需要一下几步实现即可：
    

  1. step1
    在consumer侧的resources/META-INF/dubbo目录下，创建名为com.alibaba.dubbo.rpc.cluster.Merger的文件，且内容为：
    afei=com.afei.consumer.merger.AfeiMerger
    2. 

  2. step2
    实现AfeiMerger，参考dubbo源码中若干Merger.java的实现类即可，例如：
    3. 

public class AfeiMerger implements Merger> {    private final Logger log = LoggerFactory.getLogger(this.getClass());    private static final int TOP_COUNT = 3;    /**
     * 只随机取合并后的三个结果
     */
    public List merge(List>... items) {
        List result = new ArrayList();        for (List> item : items) {            if (item != null) {
                result.addAll(item);
            }
        }

        List randList = new ArrayList();        for (int i=0; i result.size()?result.size() : TOP_COUNT;
        List finalResult = new ArrayList(resultSize);        for(int i=0; i
  1. step3
    最后一步非常简单，中配置merger="afei"即可，这个merger的值，对应step1文件内容中的key；




  1. step4
    just run it。 多运行几次，可以看到不到的结果，即达到了随机取3个结果的目的：


[{"id":101,"name":"凉拌黄瓜"},{"id":102,"name":"凉拌木耳"},{"id":2,"name":"剁椒鱼头"}]
[{"id":101,"name":"凉拌黄瓜"},{"id":3,"name":"口味虾"},{"id":2,"name":"剁椒鱼头"}]
[{"id":2,"name":"剁椒鱼头"},{"id":102,"name":"凉拌木耳"},{"id":3,"name":"口味虾"}]




作者：阿飞Javaer
链接：https://www.jianshu.com/p/512e2211f84c

来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/3271/viewspace-2802163/ ，如需转载，请注明出处，否则将追究法律责任。

0
0
上一篇： SQL Server char nchar 区别
下一篇： HTML5游戏开发（一）canvas使用
请登录后发表评论 登录
全部评论
QCHuangChao

注册时间：2019-08-30

  • 博文量
    84
  • 访问量
    673

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593