Dubbo源碼分析（四）Dubbo調用鏈-消費端（集羣容錯機制）
- Dubbo源碼分析（一）Dubbo的擴展點機制
- Dubbo源碼分析（二）Dubbo服務發佈Export
- Dubbo源碼分析（三）Dubbo的服務引用Refer
- Dubbo源碼分析（四）Dubbo調用鏈-消費端（集羣容錯機制）
- Dubbo源碼分析（五）Dubbo調用鏈-服務端
- Dubbo源碼分析（六）Dubbo通訊的編碼解碼機制
- Dubbo框架的設計細節 (未完，待續)
這篇來分析Dubbo消費端調用服務端的過程，先看一張調用鏈的總體流程圖 java
接着咱們再來看一張集羣容錯的架構圖，在集羣調用失敗時，Dubbo 提供了多種容錯方案，缺省爲 failover 重試。
源碼入口
public class Consumer {
public static void main(String[] args) {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-demo-consumer.xml"});
context.start();
//這是服務引用的源碼入口，獲取代理類
DemoService demoService = (DemoService) context.getBean("demoService"); // 獲取遠程服務代理
//這是服務調用鏈的源碼入口
String hello = demoService.sayHello("world"); // 執行遠程方法
System.out.println(hello); // 顯示調用結果
}
}
複製代碼
咱們知道，demoService是一個proxy代理類，執行demoService.sayHello方法，實際上是調用InvokerInvocationHandler.invoke方法，應該還記得proxy代理類中咱們new了一個InvokerInvocationHandler實例算法
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
String methodName = method.getName();
Class<?>[] parameterTypes = method.getParameterTypes();
···
return invoker.invoke(new RpcInvocation(method, args)).recreate();
}
複製代碼
這裏的invoker=MockClusterWrapper(FaileOverCluster)，new RpcInvocation是將全部請求參數都會轉換爲RpcInvocation，接下來咱們進入集羣部分spring
進入集羣
首先進入MockClusterWrapper.invoke方法網絡
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")) {
if (logger.isWarnEnabled()) {
logger.info("force-mock: " + invocation.getMethodName() + " force-mock enabled , url : " + directory.getUrl());
}
//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 {
if (logger.isWarnEnabled()) {
logger.info("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " + directory.getUrl(), e);
}
result = doMockInvoke(invocation, e);
}
}
}
return result;
}
複製代碼
由於咱們的配置文件中沒有配置mock，因此直接進入FaileOverCluster.invoke方法，實際上是進入父類AbstractClusterInvoker.invoke方法，看一下這個方法架構
public Result invoke(final Invocation invocation) throws RpcException {
checkWhetherDestroyed();
LoadBalance loadbalance;
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);
}
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
return doInvoke(invocation, invokers, loadbalance);
}
複製代碼
先看下list(invocation)方法app
protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
List<Invoker<T>> invokers = directory.list(invocation);
return invokers;
}
複製代碼
進入目錄查找
咱們看下directory.list(invocation)方法，這裏directory=RegistryDirectory,進入RegistryDirectory.list方法負載均衡
public List<Invoker<T>> list(Invocation invocation) throws RpcException {
···
List<Invoker<T>> invokers = doList(invocation);
List<Router> localRouters = this.routers; // local reference
if (localRouters != null && localRouters.size() > 0) {
for (Router router : localRouters) {
try {
if (router.getUrl() == null || router.getUrl().getParameter(Constants.RUNTIME_KEY, true)) {
invokers = router.route(invokers, getConsumerUrl(), invocation);
}
···
return invokers;
}
複製代碼
再進入doList方法：框架
public List<Invoker<T>> doList(Invocation invocation) {
if (forbidden) {
// 1. 沒有服務提供者 2. 服務提供者被禁用
throw new RpcException(RpcException.FORBIDDEN_EXCEPTION,
"No provider available from registry " + getUrl().getAddress() + " for service " + ··
}
List<Invoker<T>> invokers = null;
Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap; // local reference
···
return invokers == null ? new ArrayList<Invoker<T>>(0) : invokers;
}
複製代碼
從this.methodInvokerMap裏面查找一個 List<Invoker>返回dom
進入路由
接着進入路由，返回到AbstractDirectory.list方法，進入router.route()方法，此時的router=MockInvokersSelectoride
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 {
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;
}
複製代碼
進入getMockedInvokers()方法，這個方法就是將傳入的invokers和設置的路由規則匹配，得到符合條件的invokers返回
private <T> List<Invoker<T>> getNormalInvokers(final List<Invoker<T>> invokers) {
if (!hasMockProviders(invokers)) {
return invokers;
} else {
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;
}
}
複製代碼
進入負載均衡
繼續回到AbstractClusterInvoker.invoke方法，
public Result invoke(final Invocation invocation) throws RpcException {
···
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));
}
···
return doInvoke(invocation, invokers, loadbalance);
}
複製代碼
這裏先獲取loadbalance擴展點適配器LoadBalance$Adaptive，默認是RandomLoadBalance隨機負載，因此loadbalance=RandomLoadBalance，進入FailoverClusterInvoker.doInvoke方法
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
···
Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
invoked.add(invoker);
RpcContext.getContext().setInvokers((List) invoked);
try {
Result result = invoker.invoke(invocation);
if (le != null && logger.isWarnEnabled()) {
logger.warn("Although retry the method " + invocation.getMethodName()
····);
}
return result;
} catch (RpcException e) {
···
} finally {
providers.add(invoker.getUrl().getAddress());
}
}
···
}
複製代碼
進入select(loadbalance, invocation, copyinvokers, invoked)方法，最終進入RandomLoadBalance.doSelect()方法，這個隨機算法中能夠配置權重，Dubbo根據權重最終選擇一個invoker返回
遠程調用
回到 FaileOverCluster.doInvoke方法中，執行Result result = invoker.invoke(invocation);此時的invoker就是負載均衡選出來的invoker=RegistryDirectory$InvokerDelegete, 走完8個Filter，咱們進入DubboInvoker.doInvoke()方法
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(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);
}
}
複製代碼
這裏爲何DubboInvoker是個protocol? 由於RegistryDirectory.refreshInvoker.toInvokers： protocol.refer，咱們進入currentClient.request(inv, timeout).get()方法，進入HeaderExchangeChannel.request方法，進入NettyChannel.send方法，
public void send(Object message, boolean sent) throws RemotingException {
super.send(message, sent);
boolean success = true;
int timeout = 0;
try {
ChannelFuture future = channel.write(message);
if (sent) {
timeout = getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
success = future.await(timeout);
}
···
}
複製代碼
這裏最終執行ChannelFuture future = channel.write(message)，經過Netty發送網絡請求
總結
相關文章
- 1. dubbo源碼分析-集羣容錯(二)
- 2. Dubbo集羣容錯機制解析
- 3. Dubbo學習源碼總結系列四--集羣容錯機制
- 4. dubbo集羣容錯機制代碼分析1
- 5. Dubbo源碼分析之服務消費機制分析
- 6. Dubbo剖析-集羣容錯
- 7. 【dubbo源碼分析】2.dubbo消費端-ReferenceConfig初始化過程
- 8. Dubbo源碼分析（二）：Dubbo之消費端（Consumer）
- 9. Dubbo源碼分析（五）Dubbo調用鏈-服務端
- 10. DUBBO的集羣容錯機制
- 更多相關文章...
- • Swarm 集羣管理 - Docker教程
- • XML DOM 解析器錯誤 - XML DOM 教程
- • 漫談MySQL的鎖機制
- • Docker容器實戰(六) - 容器的隔離與限制
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. No provider available from registry 127.0.0.1:2181 for service com.ddbuy.ser 解決方法
- 2. Qt5.7以上調用虛擬鍵盤(支持中文)，以及源碼修改(可拖動，水平縮放)
- 3. 軟件測試面試- 購物車功能測試用例設計
- 4. ElasticSearch（概念篇）：你知道的, 爲了搜索…
- 5. redux理解
- 6. gitee創建第一個項目
- 7. 支持向量機之硬間隔（一步步推導，通俗易懂）
- 8. Mysql 異步複製延遲的原因及解決方案
- 9. 如何在運行SEPM配置嚮導時將不可認的複雜數據庫密碼改爲簡單密碼
- 10. windows系統下tftp服務器使用
歡迎關注本站公眾號,獲取更多信息
