dubbo的一次請求原始碼分析
阿新 • • 發佈：2018-12-23
呼叫某個服務首先會進入到動態代理。
InvokerInvocationHandler#invoke(Object proxy, Method method, Object[] args)
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
String methodName = method.getName();
Class<?>[] parameterTypes = method.getParameterTypes() MockClusterInvoker#invoke(Invocation invocation)
裝飾者模式，MockClusterInvoker裝飾了Invoker，主要看result = this.invoker.invoke(invocation);
public Result invoke(Invocation invocation) throws RpcException { AbstractClusterInvoker#invoke(final Invocation invocation)，預設實現FailoverClusterInvoker，首先關注list(invocation);
public Result invoke(final Invocation invocation) throws RpcException {
checkWhetherDestroyed();
LoadBalance loadbalance = null;
// binding attachments into invocation.
Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
if (contextAttachments != null && contextAttachments.size() != 0) {
((RpcInvocation) invocation).addAttachments(contextAttachments);
}
/*先往下看list方法*/
List<Invoker<T>> invokers = list(invocation);
/*獲得負載均衡的具體實現，doInvoke或用到該例項，預設RandomLoadBalance*/
if (invokers != null && !invokers.isEmpty()) {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
.getMethodParameter(RpcUtils.getMethodName(invocation), Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
}
/*跳過，預設情況下，將在非同步操作中新增呼叫ID，為了冪等*/
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
return doInvoke(invocation, invokers, loadbalance);
}
protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
List<Invoker<T>> invokers = directory.list(invocation);
return invokers;
}
directory.list(invocation);實際呼叫AbstractDirectory#list(Invocation invocation)，預設實現為RegistryDirectory
public List<Invoker<T>> list(Invocation invocation) throws RpcException {
if (destroyed) {
throw new RpcException("Directory already destroyed .url: " + getUrl());
}
/*獲得invokers，先往下看doList方法*/
List<Invoker<T>> invokers = doList(invocation);
/*獲得routers*/
List<Router> localRouters = this.routers; // local reference
if (localRouters != null && !localRouters.isEmpty()) {
/*遍歷所有Router，獲得正常的invokers*/
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) {
/*省略程式碼，log*/
}
}
}
return invokers;
}
doList(invocation);呼叫了RegistryDirectory#doList(Invocation invocation)
public List<Invoker<T>> doList(Invocation invocation) {
if (forbidden) {
// 1. No service provider 2. Service providers are disabled
/*省略程式碼，throw new RpcException*/
}
List<Invoker<T>> invokers = null;
/*快取了方法和invokers的mapping，Invoker就是具體呼叫的執行器，以後可以分析怎麼獲得的*/
Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap; // local reference
if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {
String methodName = RpcUtils.getMethodName(invocation);
/*獲得入參*/
Object[] args = RpcUtils.getArguments(invocation);
/*獲得具體的invokers*/
if (args != null && args.length > 0 && args[0] != null
&& (args[0] instanceof String || args[0].getClass().isEnum())) {
invokers = localMethodInvokerMap.get(methodName + "." + args[0]); // The routing can be enumerated according to the first parameter
}
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;
}
獲得了invokers，我再看上面的AbstractDirectory#list(Invocation invocation)方法，router.route()實際呼叫了MockInvokersSelector#route(final List<Invoker> invokers, URL url, final Invocation invocation)
public <T> List<Invoker<T>> route(final List<Invoker<T>> invokers,
URL url, final Invocation invocation) throws RpcException {
if (invocation.getAttachments() == null) {
return getNormalInvokers(invokers);
} else {
/*是否需要mock*/
String value = invocation.getAttachments().get(Constants.INVOCATION_NEED_MOCK);
if (value == null)
/*走這*/
return getNormalInvokers(invokers);
else if (Boolean.TRUE.toString().equalsIgnoreCase(value)) {
return getMockedInvokers(invokers);
}
}
return invokers;
}
private <T> List<Invoker<T>> getNormalInvokers(final List<Invoker<T>> invokers) {
/*如果沒有mock的Provider，做校驗，校驗通過返回所有invokers*/
if (!hasMockProviders(invokers)) {
return invokers;
} else {
/*否則去掉mock的Provider*/
List<Invoker<T>> sInvokers = new ArrayList<Invoker<T>>(invokers.size());
for (Invoker<T> invoker : invokers) {
if (!invoker.getUrl().getProtocol().equals(Constants.MOCK_PROTOCOL)) {
sInvokers.add(invoker);
}
}
return sInvokers;
}
}
AbstractDirectory#list(Invocation invocation)方法終於結束了，主要就是獲得了正常的invokers。
小結：首先Directory獲得所有Invokers，然後Router獲得所有非mock的Invokers。
接著回到AbstractClusterInvoker#invoke(final Invocation invocation)，獲得具體負載均衡的例項後，呼叫了FailoverClusterInvoker#doInvoke(Invocation invocation, final List<Invoker> invokers, LoadBalance loadbalance)，下面這部分主要為負載到某個Invoker，不想看的可以直接跳到invoke
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
List<Invoker<T>> copyinvokers = invokers;
/*檢查是否為空*/
checkInvokers(copyinvokers, invocation);
int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
if (len <= 0) {
len = 1;
}
// retry loop.
RpcException le = null; // last exception.
List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size()); // invoked invokers.
Set<String> providers = new HashSet<String>(len);
for (int i = 0; i < len; i++) {
//在重試之前重新選擇以避免Invokers發生變化。
//注意：如果`invokers`改變了，那麼`invoked`也會失去準確性。
if (i > 0) {
checkWhetherDestroyed();
copyinvokers = list(invocation);
// check again
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()) {
/*省略程式碼，log.warn*/
}
return result;
} catch (RpcException e) {
if (e.isBiz()) { // biz exception.
throw e;
}
le = e;
} catch (Throwable e) {
le = new RpcException(e.getMessage(), e);
} finally {
providers.add(invoker.getUrl().getAddress());
}
}
/*省略程式碼，throw new RpcException*/
}
AbstractClusterInvoker#select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected)
protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.isEmpty())
return null;
String methodName = invocation == null ? "" : invocation.getMethodName();
boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName, Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY);
{
//ignore overloaded method
if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
stickyInvoker = null;
}
//ignore concurrency problem
if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
if (availablecheck && stickyInvoker.isAvailable()) {
return stickyInvoker;
}
}
}
Invoker<T> invoker = doSelect(loadbalance, invocation, invokers, selected);
if (sticky) {
stickyInvoker = invoker;
}
return invoker;
}
AbstractClusterInvoker#doSelect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected)
private Invoker<T> doSelect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.isEmpty())
return null;
/*可有一個直接返回*/
if (invokers.size() == 1)
return invokers.get(0);
if (loadbalance == null) {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
}
/*之前例項化的LoadBalance*/
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 {
//檢查當前所選呼叫者的索引，如果不是