摘要：源碼分析一創(chuàng)建該類(lèi)實(shí)現(xiàn)了接口，是分組集合的集群實(shí)現(xiàn)。三工廠類(lèi)，獲得指定類(lèi)型的對(duì)象。后記該部分相關(guān)的源碼解析地址該文章講解了集群中關(guān)于分組聚合實(shí)現(xiàn)的部分。接下來(lái)我將開(kāi)始對(duì)集群模塊關(guān)于路由部分進(jìn)行講解。

目標(biāo)：介紹dubbo中集群的分組聚合，介紹dubbo-cluster下merger包的源碼。

按組合并返回結(jié)果 ，比如菜單服務(wù)，接口一樣，但有多種實(shí)現(xiàn)，用group區(qū)分，現(xiàn)在消費(fèi)方需從每種group中調(diào)用一次返回結(jié)果，合并結(jié)果返回，這樣就可以實(shí)現(xiàn)聚合菜單項(xiàng)。這個(gè)時(shí)候就要用到分組聚合。

public class MergeableCluster implements Cluster {

    public static final String NAME = "mergeable";

    @Override
    public  Invoker join(Directory directory) throws RpcException {
        // 創(chuàng)建MergeableClusterInvoker
        return new MergeableClusterInvoker(directory);
    }

}

該類(lèi)實(shí)現(xiàn)了Cluster接口，是分組集合的集群實(shí)現(xiàn)。

該類(lèi)是分組聚合的實(shí)現(xiàn)類(lèi)，其中最關(guān)機(jī)的就是invoke方法。

@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);
    // 如果沒(méi)有設(shè)置需要聚合，則只調(diào)用一個(gè)invoker的
    if (ConfigUtils.isEmpty(merger)) { // If a method doesn"t have a merger, only invoke one Group
        // 只要有一個(gè)可用就返回
        for (final Invoker invoker : invokers) {
            if (invoker.isAvailable()) {
                return invoker.invoke(invocation);
            }
        }
        return invokers.iterator().next().invoke(invocation);
    }

    // 返回類(lèi)型
    Class returnType;
    try {
        // 獲得返回類(lèi)型
        returnType = getInterface().getMethod(
                invocation.getMethodName(), invocation.getParameterTypes()).getReturnType();
    } catch (NoSuchMethodException e) {
        returnType = null;
    }

    // 結(jié)果集合
    Map> results = new HashMap>();
    // 循環(huán)invokers
    for (final Invoker invoker : invokers) {
        // 獲得每次調(diào)用的future
        Future future = executor.submit(new Callable() {
            @Override
            public Result call() throws Exception {
                // 回調(diào)，把返回結(jié)果放入future
                return invoker.invoke(new RpcInvocation(invocation, invoker));
            }
        });
        // 加入集合
        results.put(invoker.getUrl().getServiceKey(), future);
    }

    Object result = null;

    List resultList = new ArrayList(results.size());

    // 獲得超時(shí)時(shí)間
    int timeout = getUrl().getMethodParameter(invocation.getMethodName(), Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
    // 遍歷每一個(gè)結(jié)果
    for (Map.Entry> entry : results.entrySet()) {
        Future future = entry.getValue();
        try {
            // 獲得調(diào)用返回的結(jié)果
            Result r = future.get(timeout, TimeUnit.MILLISECONDS);
            if (r.hasException()) {
                log.error("Invoke " + getGroupDescFromServiceKey(entry.getKey()) + 
                                " failed: " + r.getException().getMessage(), 
                        r.getException());
            } else {
                // 加入集合
                resultList.add(r);
            }
        } catch (Exception e) {
            throw new RpcException("Failed to invoke service " + entry.getKey() + ": " + e.getMessage(), e);
        }
    }

    // 如果為空，則返回空的結(jié)果
    if (resultList.isEmpty()) {
        return new RpcResult((Object) null);
    } else if (resultList.size() == 1) {
        // 如果只有一個(gè)結(jié)果，則返回該結(jié)果
        return resultList.iterator().next();
    }

    // 如果返回類(lèi)型是void，也就是沒(méi)有返回值，那么返回空結(jié)果
    if (returnType == void.class) {
        return new RpcResult((Object) null);
    }

    // 根據(jù)方法來(lái)合并，將調(diào)用返回結(jié)果的指定方法進(jìn)行合并
    if (merger.startsWith(".")) {
        merger = merger.substring(1);
        Method method;
        try {
            // 獲得方法
            method = returnType.getMethod(merger, returnType);
        } catch (NoSuchMethodException e) {
            throw new RpcException("Can not merge result because missing method [ " + merger + " ] in class [ " + 
                    returnType.getClass().getName() + " ]");
        }

        // 有 Method ，進(jìn)行合并
        if (!Modifier.isPublic(method.getModifiers())) {
            method.setAccessible(true);
        }
        // 從集合中移除
        result = resultList.remove(0).getValue();
        try {
            // 方法返回類(lèi)型匹配，合并時(shí)，修改 result
            if (method.getReturnType() != void.class
                    && method.getReturnType().isAssignableFrom(result.getClass())) {
                for (Result r : resultList) {
                    result = method.invoke(result, r.getValue());
                }
            } else {
                // 方法返回類(lèi)型不匹配，合并時(shí)，不修改 result
                for (Result r : resultList) {
                    method.invoke(result, r.getValue());
                }
            }
        } catch (Exception e) {
            throw new RpcException("Can not merge result: " + e.getMessage(), e);
        }
    } else {
        // 基于 Merger
        Merger resultMerger;
        // 如果是默認(rèn)的方式
        if (ConfigUtils.isDefault(merger)) {
            // 獲得該類(lèi)型的合并方式
            resultMerger = MergerFactory.getMerger(returnType);
        } else {
            // 如果不是默認(rèn)的，則配置中指定獲得Merger的實(shí)現(xiàn)類(lèi)
            resultMerger = ExtensionLoader.getExtensionLoader(Merger.class).getExtension(merger);
        }
        if (resultMerger != null) {
            List