閱讀須知
- dubbo版本：2.6.0
- spring版本：4.3.8
- 文章中使用注釋的方法會做深入分析
正文
在分析consumer初始化時，我們看到了關聯服務引用建立代理的過程，最終會調用JavassistProxyFactory的getProxy方法來建立代理，并用InvokerInvocationHandler對Invoker進行了包裝，InvokerInvocationHandler實作了JDK的InvocationHandler，這個接口相信熟悉JDK動态代理的同學一定不陌生，是以我們在調用服務的方法時就會調用其invoke方法，我們來看實作：
InvokerInvocationHandler：
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
String methodName = method.getName();
Class<?>[] parameterTypes = method.getParameterTypes();
if (method.getDeclaringClass() == Object.class) {
return method.invoke(invoker, args);
}
if ("toString".equals(methodName) && parameterTypes.length == 0) {
return invoker.toString();
}
if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
return invoker.hashCode();
}
if ("equals".equals(methodName) && parameterTypes.length == 1) {
return invoker.equals(args[0]);
}
/* 将方法和參數封裝到RpcInvocation中，調用Invoker的invoke方法 */
return invoker.invoke(new RpcInvocation(method, args)).recreate();
}
這裡的Invoker我們在也分析consumer初始化時同樣看到過，是由FailoverCluster的FailoverClusterInvoker：
AbstractClusterInvoker：
public Result invoke(final Invocation invocation) throws RpcException {
checkWhetherDestroyed(); // 檢查invoker是否已經銷毀
LoadBalance loadbalance;
/* 擷取invoker集合 */
List<Invoker<T>> invokers = list(invocation);
if (invokers != null && invokers.size() > 0) {
// 擷取負載均衡政策，預設為随機
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
.getMethodParameter(invocation.getMethodName(), Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
} else {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
}
// 幂等操作：預設情況下，将在異步操作中添加調用ID
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
/* 調用 */
return doInvoke(invocation, invokers, loadbalance);
}
AbstractClusterInvoker：
protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
/* 這裡的directory在RegistryProtocol的doRefer方法中建構Invoker時傳入，為RegistryDirectory */
List<Invoker<T>> invokers = directory.list(invocation);
return invokers;
}
AbstractDirectory：
public List<Invoker<T>> list(Invocation invocation) throws RpcException {
if (destroyed) {
throw new RpcException("Directory already destroyed .url: " + getUrl());
}
/* 擷取invoker */
List<Invoker<T>> invokers = doList(invocation);
// 路由處理，筆者的環境中Router集合中隻有MockInvokersSelector，用來處理mock服務
List<Router> localRouters = this.routers;
if (localRouters != null && localRouters.size() > 0) {
for (Router router : localRouters) {
try {
if (router.getUrl() == null || router.getUrl().getParameter(Constants.RUNTIME_KEY, false)) {
invokers = router.route(invokers, getConsumerUrl(), invocation);
}
} catch (Throwable t) {
logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
}
}
}
return invokers;
}
RegistryDirectory：
public List<Invoker<T>> doList(Invocation invocation) {
if (forbidden) {
throw new RpcException(RpcException.FORBIDDEN_EXCEPTION,
"No provider available from registry " + getUrl().getAddress() + " for service " + getConsumerUrl().getServiceKey() + " on consumer " + NetUtils.getLocalHost()
+ " use dubbo version " + Version.getVersion() + ", may be providers disabled or not registered ?");
}
List<Invoker<T>> invokers = null;
// 這裡的methodInvokerMap在consumer初始化時訂閱注冊中心的providers、configuration等相關資訊時收到通知時初始化
Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap;
if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {
String methodName = RpcUtils.getMethodName(invocation);
Object[] args = RpcUtils.getArguments(invocation);
// 依次采用不同的方式從map中擷取invoker
if (args != null && args.length > 0 && args[0] != null
&& (args[0] instanceof String || args[0].getClass().isEnum())) {
invokers = localMethodInvokerMap.get(methodName + "." + args[0]);
}
if (invokers == null) {
invokers = localMethodInvokerMap.get(methodName);
}
if (invokers == null) {
invokers = localMethodInvokerMap.get(Constants.ANY_VALUE);
}
if (invokers == null) {
Iterator<List<Invoker<T>>> iterator = localMethodInvokerMap.values().iterator();
if (iterator.hasNext()) {
invokers = iterator.next();
}
}
}
return invokers == null ? new ArrayList<Invoker<T>>(0) : invokers;
}
FailoverClusterInvoker：
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
List<Invoker<T>> copyinvokers = invokers;
checkInvokers(copyinvokers, invocation);
// 重試次數，第一次調用不算重試，是以加1
int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
if (len <= 0) {
len = 1;
}
RpcException le = null;
// 存儲已經調用過的invoker
List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size());
Set<String> providers = new HashSet<String>(len);
// 重試循環
for (int i = 0; i < len; i++) {
// 在重試之前重新選擇以避免候選invokers的更改
// 注意：如果invokers改變了，那麼invoked集合也會失去準确性
if (i > 0) {
// 重複第一次調用的幾個步驟
checkWhetherDestroyed();
copyinvokers = list(invocation);
checkInvokers(copyinvokers, invocation);
}
/* 選擇一個invoker */
Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
invoked.add(invoker);
RpcContext.getContext().setInvokers((List) invoked);
try {
/* 調用invoke方法傳回結果 */
Result result = invoker.invoke(invocation);
if (le != null && logger.isWarnEnabled()) {
logger.warn("Although retry the method " + invocation.getMethodName()
+ " in the service " + getInterface().getName()
+ " was successful by the provider " + invoker.getUrl().getAddress()
+ ", but there have been failed providers " + providers
+ " (" + providers.size() + "/" + copyinvokers.size()
+ ") from the registry " + directory.getUrl().getAddress()
+ " on the consumer " + NetUtils.getLocalHost()
+ " using the dubbo version " + Version.getVersion() + ". Last error is: "
+ le.getMessage(), le);
}
return result;
} catch (RpcException e) {
if (e.isBiz()) {
throw e;
}
le = e;
} catch (Throwable e) {
le = new RpcException(e.getMessage(), e);
} finally {
providers.add(invoker.getUrl().getAddress());
}
}
throw new RpcException(le != null ? le.getCode() : 0, "Failed to invoke the method "
+ invocation.getMethodName() + " in the service " + getInterface().getName()
+ ". Tried " + len + " times of the providers " + providers
+ " (" + providers.size() + "/" + copyinvokers.size()
+ ") from the registry " + directory.getUrl().getAddress()
+ " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version "
+ Version.getVersion() + ". Last error is: "
+ (le != null ? le.getMessage() : ""), le != null && le.getCause() != null ? le.getCause() : le);
}
AbstractClusterInvoker：
protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.size() == 0)
return null;
String methodName = invocation == null ? "" : invocation.getMethodName();
boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName, Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY);
{
// 忽略重載方法
if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
stickyInvoker = null;
}
// 忽略并發問題
if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
if (availablecheck && stickyInvoker.isAvailable()) {
return stickyInvoker;
}
}
}
/* 使用負載均衡政策選擇invoker */
Invoker<T> invoker = doselect(loadbalance, invocation, invokers, selected);
if (sticky) {
stickyInvoker = invoker;
}
return invoker;
}
AbstractClusterInvoker：
private Invoker<T> doselect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.size() == 0)
return null;
if (invokers.size() == 1)
return invokers.get(0); // 隻有一個invoker直接傳回
// 如果隻有兩個invoker，使用輪詢政策
if (invokers.size() == 2 && selected != null && selected.size() > 0) {
return selected.get(0) == invokers.get(0) ? invokers.get(1) : invokers.get(0);
}
// 負載均衡政策選擇invoker
Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);
// 如果invoker在selected中或者invoker不可用并且availablecheck為true，則重新選擇
if ((selected != null && selected.contains(invoker))
|| (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
try {
/* 重新選擇 */
Invoker<T> rinvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
if (rinvoker != null) {
invoker = rinvoker;
} else {
int index = invokers.indexOf(invoker);
try {
// 檢查目前所選invoker的index，如果不是最後一個，請選擇索引為index + 1的invoker
invoker = index < invokers.size() - 1 ? invokers.get(index + 1) : invoker;
} catch (Exception e) {
logger.warn(e.getMessage() + " may because invokers list dynamic change, ignore.", e);
}
}
} catch (Throwable t) {
logger.error("clustor relselect fail reason is :" + t.getMessage() + " if can not slove ,you can set cluster.availablecheck=false in url", t);
}
}
return invoker;
}
關于dubbo的負載均衡政策，我們會用單獨的文章進行分析。
AbstractClusterInvoker：
private Invoker<T> reselect(LoadBalance loadbalance, Invocation invocation,
List<Invoker<T>> invokers, List<Invoker<T>> selected, boolean availablecheck)
throws RpcException {
// 事先配置設定一個清單，肯定會使用此清單
List<Invoker<T>> reselectInvokers = new ArrayList<Invoker<T>>(invokers.size() > 1 ? (invokers.size() - 1) : invokers.size());
// 首先嘗試從未在selected中的invoker中選擇一個，
if (availablecheck) {
for (Invoker<T> invoker : invokers) {
// invoker.isAvailable() 需要被檢查
if (invoker.isAvailable()) {
if (selected == null || !selected.contains(invoker)) {
reselectInvokers.add(invoker);
}
}
}
if (reselectInvokers.size() > 0) {
return loadbalance.select(reselectInvokers, getUrl(), invocation);
}
} else {
// 不檢查invoker.isAvailable()
for (Invoker<T> invoker : invokers) {
if (selected == null || !selected.contains(invoker)) {
reselectInvokers.add(invoker);
}
}
if (reselectInvokers.size() > 0) {
return loadbalance.select(reselectInvokers, getUrl(), invocation);
}
}
// 隻需使用loadbalance政策選擇一個可用的invoker
{
if (selected != null) {
for (Invoker<T> invoker : selected) {
// available優先
if ((invoker.isAvailable())
&& !reselectInvokers.contains(invoker)) {
reselectInvokers.add(invoker);
}
}
}
if (reselectInvokers.size() > 0) {
return loadbalance.select(reselectInvokers, getUrl(), invocation);
}
}
return null;
}
選擇好invoker後，接下來就是調用其invoke方法，這裡的invoker是在訂閱注冊中心的相關資訊後，收到通知時建立的用于包裝invoker的包裝類（詳見Dubbo源碼解析之consumer關聯provider），是RegistryDirectory的内部類InvokerDelegate。
InvokerWrapper：
public Result invoke(Invocation invocation) throws RpcException {
return invoker.invoke(invocation); // DubboInvoker
}
AbstractInvoker：
public Result invoke(Invocation inv) throws RpcException {
if (destroyed.get()) {
throw new RpcException("Rpc invoker for service " + this + " on consumer " + NetUtils.getLocalHost()
+ " use dubbo version " + Version.getVersion()
+ " is DESTROYED, can not be invoked any more!");
}
RpcInvocation invocation = (RpcInvocation) inv;
invocation.setInvoker(this);
if (attachment != null && attachment.size() > 0) {
invocation.addAttachmentsIfAbsent(attachment);
}
Map<String, String> context = RpcContext.getContext().getAttachments();
if (context != null) {
invocation.addAttachmentsIfAbsent(context);
}
if (getUrl().getMethodParameter(invocation.getMethodName(), Constants.ASYNC_KEY, false)) {
invocation.setAttachment(Constants.ASYNC_KEY, Boolean.TRUE.toString());
}
// 幂等操作：預設情況下，将在異步操作中添加調用ID
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
try {
/* 調用 */
return doInvoke(invocation);
} catch (InvocationTargetException e) {
Throwable te = e.getTargetException();
if (te == null) {
return new RpcResult(e);
} else {
if (te instanceof RpcException) {
((RpcException) te).setCode(RpcException.BIZ_EXCEPTION);
}
return new RpcResult(te);
}
} catch (RpcException e) {
if (e.isBiz()) {
return new RpcResult(e);
} else {
throw e;
}
} catch (Throwable e) {
return new RpcResult(e);
}
}
DubboInvoker：
protected Result doInvoke(final Invocation invocation) throws Throwable {
RpcInvocation inv = (RpcInvocation) invocation;
final String methodName = RpcUtils.getMethodName(invocation);
inv.setAttachment(Constants.PATH_KEY, getUrl().getPath());
inv.setAttachment(Constants.VERSION_KEY, version);
// 擷取client，輪詢
ExchangeClient currentClient;
if (clients.length == 1) {
currentClient = clients[0];
} else {
currentClient = clients[index.getAndIncrement() % clients.length];
}
try {
boolean isAsync = RpcUtils.isAsync(getUrl(), invocation); // 是否異步
boolean isOneway = RpcUtils.isOneway(getUrl(), invocation); // 是否單雙工
int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
if (isOneway) {
boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
currentClient.send(inv, isSent); // 直接發送請求
RpcContext.getContext().setFuture(null);
return new RpcResult();
} else if (isAsync) {
// 異步采用future模式
ResponseFuture future = currentClient.request(inv, timeout);
RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
return new RpcResult();
} else {
RpcContext.getContext().setFuture(null);
return (Result) currentClient.request(inv, timeout).get();
}
} catch (TimeoutException e) {
throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
} catch (RemotingException e) {
throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
}
}
這裡發送請求的過程我們在分析provider調用過程源碼的時候分析過，這裡複用的是同樣的流程，到這裡，consumer調用過程就完成了。
![實驗吧WEBWP(一）[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)