dubbo的一次請求源碼分析
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);
}
/*object#method mock*/
if ("toString".equals(methodName) && parameterTypes.length == 0) {
return invoker.toString();
}
/*...*/
/*省略hashCode和equals代碼*/
/*其餘實現方法invoke*/
return invoker.invoke(new RpcInvocation(method, args)).recreate();
}
複製代碼
MockClusterInvoker#invoke(Invocation invocation) 裝飾者模式，MockClusterInvoker裝飾了Invoker，主要看result = this.invoker.invoke(invocation);java
public Result invoke(Invocation invocation) throws RpcException {
Result result = null;
String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), Constants.MOCK_KEY, Boolean.FALSE.toString()).trim();
if (value.length() == 0 || value.equalsIgnoreCase("false")) {
//no mock
result = this.invoker.invoke(invocation);
} else if (value.startsWith("force")) {
/*省略代碼，log*/
//force:direct mock
result = doMockInvoke(invocation, null);
} else {
//fail-mock
try {
result = this.invoker.invoke(invocation);
} catch (RpcException e) {
if (e.isBiz()) {
throw e;
} else {
/*省略代碼，log*/
result = doMockInvoke(invocation, e);
}
}
}
return result;
}
複製代碼
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)，缺省實現爲RegistryDirectoryapp
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)dom
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，不想看的能夠直接跳到invokeide
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)oop
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)ui
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 {
//檢查當前所選調用者的索引，若是不是最後一個，選擇index+1的這個。
int index = invokers.indexOf(invoker);
try {
//Avoid collision
invoker = index < invokers.size() - 1 ? invokers.get(index + 1) : invokers.get(0);
} catch (Exception e) {
/*省略代碼，log.warn*/
}
}
} catch (Throwable t) {
/*省略代碼，log.error*/
}
}
return invoker;
}
複製代碼
AbstractLoadBalance#select(List<Invoker<T>> invokers, URL url, Invocation invocation)this
public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
if (invokers == null || invokers.isEmpty())
return null;
if (invokers.size() == 1)
return invokers.get(0);
return doSelect(invokers, url, invocation);
}
複製代碼
RandomLoadBalance#doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation)，計算隨機值，判斷在哪一個權重範圍內，則返回這個範圍中的Invoker
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
int length = invokers.size(); //invokers總數
int totalWeight = 0; //總權重
boolean sameWeight = true; // 權重是都否同樣
for (int i = 0; i < length; i++) {
int weight = getWeight(invokers.get(i), invocation);
totalWeight += weight; // 累計總權重
if (sameWeight && i > 0
&& weight != getWeight(invokers.get(i - 1), invocation)) {
sameWeight = false; //計算全部權重是否同樣
}
}
if (totalWeight > 0 && !sameWeight) {
// 若是（並不是每一個調用者具備相同的權重而且至少一個調用者的權重>0），則根據totalWeight隨機選擇。
int offset = random.nextInt(totalWeight);
// 根據隨機值返回一個調用者。
for (int i = 0; i < length; i++) {
offset -= getWeight(invokers.get(i), invocation);
if (offset < 0) {
return invokers.get(i);
}
}
}
//若是全部調用者具備相同的權重值或totalWeight = 0，則均勻返回。
return invokers.get(random.nextInt(length));
}
複製代碼
Invoker獲取到了就能夠執行了，回到FailoverClusterInvoker#doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance)，接下來invoker.invoke(invocation)，該方法通過了幾層裝飾，調用責任鏈（之後再寫怎麼生成噠），最後調用了AbstractInvoker(實現類DubboInvoker)#invoke(Invocation inv)
public Result invoke(Invocation inv) throws RpcException {
if (destroyed.get()) {
/*省略代碼，throw new RpcException*/
}
RpcInvocation invocation = (RpcInvocation) inv;
invocation.setInvoker(this);
if (attachment != null && attachment.size() > 0) {
invocation.addAttachmentsIfAbsent(attachment);
}
Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
if (contextAttachments != null && contextAttachments.size() != 0) {
/** * invocation.addAttachmentsIfAbsent(context){@link RpcInvocation#addAttachmentsIfAbsent(Map)}should not be used here, * because the {@link RpcContext#setAttachment(String, String)} is passed in the Filter when the call is triggered * by the built-in retry mechanism of the Dubbo. The attachment to update RpcContext will no longer work, which is * a mistake in most cases (for example, through Filter to RpcContext output traceId and spanId and other information). */
invocation.addAttachments(contextAttachments);
}
if (getUrl().getMethodParameter(invocation.getMethodName(), Constants.ASYNC_KEY, false)) {
invocation.setAttachment(Constants.ASYNC_KEY, Boolean.TRUE.toString());
}
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
try {
return doInvoke(invocation);
} catch (InvocationTargetException e) { // biz exception
/*省略，異常處理*/
} catch (RpcException e) {
/*省略，異常處理*/
} catch (Throwable e) {
return new RpcResult(e);
}
}
複製代碼
DubboInvoker.doInvoke(final Invocation invocation)，發送接收消息
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);
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) {
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*/
} catch (RemotingException e) {
/*省略代碼，throw new RpcException*/
}
}
複製代碼
總結：
- 首先經過Directory得到全部Invokers；
- 接着Router得到全部非mock的Invokers；
- 而後經過LoadBalance得到某個Invoker；
- 最後通過一些裝飾和責任鏈後，調用發送接收消息，獲得最後結果。
相關文章
- 1. dubbo的一次請求源碼分析
- 2. spring mvc 5.1.1.RELEASE的一次請求過程源碼分析
- 3. OkHttp 源碼分析（一）—— 請求流程
- 4. dubbo源碼分析（一）
- 5. 真*dubbo 源碼分析(一)
- 6. Dubbo源碼分析之一
- 7. dubbo源碼分析一：整體分析
- 8. dubbo源碼分析一：總體分析
- 9. dubbo服務端處理請求源碼分析
- 10. dubbo源碼分析系列——dubbo的SPI機制源碼分析
- 更多相關文章...
- • HTTP 請求方法 - HTTP 教程
- • 發送ICMP時間戳請求 - TCP/IP教程
- • 互聯網組織的未來：剖析GitHub員工的任性之源
- • Java Agent入門實戰（二）-Instrumentation源碼概述
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
