下面我們來說一下Dubbo的Merger實(shí)現(xiàn)万细。在開發(fā)中粪般，有這么一種情況尝哆，先定義了一個UserService接口频蛔，有UserServiceImpl和CategoryUserServiceImpl兩種實(shí)現(xiàn)，它們又分別屬于user和category兩個組照雁，consumer將調(diào)用這兩個服務(wù)蚕愤，并按照自定義策略合并返回結(jié)果，作為最終結(jié)果饺蚊。這就需要Dubbo的Merger來實(shí)現(xiàn)了萍诱。我們先來看一下MergeableClusterInvoker的invoke方法
public Result invoke(final Invocation invocation) throws RpcException {
// 獲得 Invoker 集合
List<Invoker<T>> invokers = directory.list(invocation);
// 獲得 Merger 拓展名
String merger = getUrl().getMethodParameter(invocation.getMethodName(), Constants.MERGER_KEY);
// 若果未配置拓展，直接調(diào)用首個可用的 Invoker 對象
if (ConfigUtils.isEmpty(merger)) { // If a method doesn't have a merger, only invoke one Group
for (final Invoker<T> invoker : invokers) {
if (invoker.isAvailable()) {
return invoker.invoke(invocation);
}
}
return invokers.iterator().next().invoke(invocation);
}
// 通過反射污呼，獲得返回類型
Class<?> returnType;
try {
returnType = getInterface().getMethod(invocation.getMethodName(), invocation.getParameterTypes()).getReturnType();
} catch (NoSuchMethodException e) {
returnType = null;
}
// 提交線程池裕坊，并行執(zhí)行，發(fā)起 RPC 調(diào)用曙求，并添加到 results 中
Map<String, Future<Result>> results = new HashMap<String, Future<Result>>();
for (final Invoker<T> invoker : invokers) {
Future<Result> future = executor.submit(new Callable<Result>() {
public Result call() {
// RPC 調(diào)用
return invoker.invoke(new RpcInvocation(invocation, invoker));
}
});
results.put(invoker.getUrl().getServiceKey(), future);
}
// 阻塞等待執(zhí)行執(zhí)行結(jié)果碍庵，并添加到 resultList 中
List<Result> resultList = new ArrayList<Result>(results.size());
int timeout = getUrl().getMethodParameter(invocation.getMethodName(), Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
for (Map.Entry<String, Future<Result>> entry : results.entrySet()) {
Future<Result> future = entry.getValue();
try {
Result r = future.get(timeout, TimeUnit.MILLISECONDS);
if (r.hasException()) { // 異常 Result 映企，打印錯誤日志悟狱，忽略
log.error(new StringBuilder(32).append("Invoke ").append(getGroupDescFromServiceKey(entry.getKey())).append(" failed: ").append(r.getException().getMessage()).toString(), r.getException());
} else { // 正常 Result ，添加到 resultList 中
resultList.add(r);
}
} catch (Exception e) { // 異常堰氓，拋出 RpcException 異常
throw new RpcException(new StringBuilder(32).append("Failed to invoke service ").append(entry.getKey()).append(": ").append(e.getMessage()).toString(), e);
}
}
// 結(jié)果大小為空挤渐，返回空的 RpcResult
if (resultList.isEmpty()) {
return new RpcResult((Object) null);
// 結(jié)果大小為 1 ，返回首個 RpcResult
} else if (resultList.size() == 1) {
return resultList.iterator().next();
}
// 返回類型為 void 双絮，返回空的 RpcResult
if (returnType == void.class) {
return new RpcResult((Object) null);
}
Object result;
// 【第 1 種】基于合并方法
if (merger.startsWith(".")) {
// 獲得合并方法 Method
merger = merger.substring(1);
Method method;
try {
method = returnType.getMethod(merger, returnType);
} catch (NoSuchMethodException e) {
throw new RpcException(new StringBuilder(32).append("Can not merge result because missing method [ ").append(merger).append(" ] in class [ ").append(returnType.getClass().getName()).append(" ]").toString());
}
// 有 Method 浴麻，進(jìn)行合并
if (method != null) {
if (!Modifier.isPublic(method.getModifiers())) {
method.setAccessible(true);
}
result = resultList.remove(0).getValue();
try {
// 方法返回類型匹配得问，合并時，修改 result
if (method.getReturnType() != void.class && method.getReturnType().isAssignableFrom(result.getClass())) {
for (Result r : resultList) {
result = method.invoke(result, r.getValue());
}
// 方法返回類型不匹配软免，合并時宫纬，不修改 result
} else {
for (Result r : resultList) {
method.invoke(result, r.getValue());
}
}
} catch (Exception e) {
throw new RpcException(new StringBuilder(32).append("Can not merge result: ").append(e.getMessage()).toString(), e);
}
// 無 Method ，拋出 RpcException 異常
} else {
throw new RpcException(new StringBuilder(32).append("Can not merge result because missing method [ ").append(merger).append(" ] in class [ ").append(returnType.getClass().getName()).append(" ]").toString());
}
// 【第 2 種】基于 Merger
} else {
Merger resultMerger;
// 【第 2.1 種】根據(jù)返回值類型自動匹配 Merger
if (ConfigUtils.isDefault(merger)) {
resultMerger = MergerFactory.getMerger(returnType);
// 【第 2.2 種】指定 Merger
} else {
resultMerger = ExtensionLoader.getExtensionLoader(Merger.class).getExtension(merger);
}
// 有 Merger 膏萧，進(jìn)行合并
if (resultMerger != null) {
List<Object> rets = new ArrayList<Object>(resultList.size());
for (Result r : resultList) {
rets.add(r.getValue());
}
result = resultMerger.merge(rets.toArray((Object[]) Array.newInstance(returnType, 0)));
// 無 Merger 漓骚，拋出 RpcException 異常
} else {
throw new RpcException("There is no merger to merge result.");
}
}
// 返回 RpcResult 結(jié)果
return new RpcResult(result);
}
調(diào)用了Merger進(jìn)行結(jié)果的合并處理。我們再來看一個類MergerFactory榛泛，這個類是生產(chǎn)Merger的工廠類
public static <T> Merger<T> getMerger(Class<T> returnType) {
Merger result;
// 數(shù)組類型
if (returnType.isArray()) {
Class type = returnType.getComponentType();
// 從緩存中獲得 Merger 對象
result = mergerCache.get(type);
if (result == null) {
loadMergers();
result = mergerCache.get(type);
}
// 獲取不到蝌蹂，使用 ArrayMerger
if (result == null && !type.isPrimitive()) {
result = ArrayMerger.INSTANCE;
}
// 普通類型
} else {
// 從緩存中獲得 Merger 對象
result = mergerCache.get(returnType);
if (result == null) {
loadMergers();
result = mergerCache.get(returnType);
}
}
return result;
}
我們再來看一個類ArrayMerger，這個類是數(shù)組結(jié)果合并的處理類曹锨，實(shí)現(xiàn)了merge方法
if (others.length == 0) {
return null;
}
int totalLen = 0;
for (int i = 0; i < others.length; i++) {
Object item = others[i];
if (item != null && item.getClass().isArray()) {
totalLen += Array.getLength(item);
} else {
throw new IllegalArgumentException(
new StringBuilder(32).append(i + 1)
.append("th argument is not an array").toString());
}
}
if (totalLen == 0) {
return null;
}
Class<?> type = others[0].getClass().getComponentType();
Object result = Array.newInstance(type, totalLen);
int index = 0;
for (Object array : others) {
for (int i = 0; i < Array.getLength(array); i++) {
Array.set(result, index++, Array.get(array, i));
}
}
return (Object[]) result;
再來看一個類MapMerger孤个，這個類是Map結(jié)果合并的處理類
if (items.length == 0) {
return null;
}
// 創(chuàng)建結(jié)果 Map
Map<Object, Object> result = new HashMap<Object, Object>();
// 合并多個 Map
for (Map<?, ?> item : items) {
if (item != null) {
result.putAll(item);
}
}
return result;
還有很多數(shù)據(jù)類型的處理類，比如BooleanArrayMerger,ByteArrayMerger,CharArrayMerger等沛简，就不在這里具體說了齐鲤。
通過Merger實(shí)現(xiàn)了對多個返回結(jié)果的處理。
Dubbo的Merger機(jī)制就介紹到這里了椒楣。
