2. 程式人生 > >dubbo 服務呼叫原始碼分析

dubbo 服務呼叫原始碼分析

阿新 • • 發佈：2019-01-15

下面是呼叫方法棧和核心程式碼分析

InvokerInvocationHandler.invoke(Object, Method, Object[]) line: 38

com.alibaba.dubbo.rpc.cluster.support.wrapper.MockClusterInvoker.invoke(Invocation)
com.alibaba.dubbo.rpc.cluster.support.AbstractClusterInvoker.invoke(Invocation)






 
public Result invoke(final Invocation invocation) throws RpcException {


        checkWheatherDestoried();


        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);
        }
//冪等操作:非同步操作預設新增invocation id
        RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
        return doInvoke(invocation, invokers, loadbalance);
    }





com.alibaba.dubbo.rpc.cluster.support.FailoverClusterInvoker.doInvoke(Invocation, List<Invoker<T>>, LoadBalance)






  @SuppressWarnings({ "unchecked", "rawtypes" })
    public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
    List<Invoker<T>> copyinvokers = invokers;
    checkInvokers(copyinvokers, invocation);
        //獲取重試次數，dubbo有容錯機制
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++) {
        //重試時，進行重新選擇，避免重試時invoker列表已發生變化.
        //注意：如果列表發生了變化，那麼invoked判斷會失效，因為invoker示例已經改變
        if (i > 0) {
        checkWheatherDestoried();
        copyinvokers = list(invocation);
        //重新檢查一下
        checkInvokers(copyinvokers, invocation);
        }
//使用loadbalance選擇invoker.
            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()
                            + " 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()) { // biz exception.
                    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);
    }



com.alibaba.dubbo.rpc.protocol.InvokerWrapper.invoke(Invocation)

com.alibaba.dubbo.rpc.filter.ConsumerContextFilter.invoke(Invoker<?>, Invocation)

com.alibaba.dubbo.rpc.protocol.dubbo.filter.FutureFilter.invoke(Invoker<?>, Invocation)

com.alibaba.dubbo.rpc.protocol.dubbo.DubboInvoker.doInvoke(Invocation)
@Override
    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);
        }
    }

com.alibaba.dubbo.remoting.transport.AbstractClient.send(Object, boolean)

public void send(Object message, boolean sent) throws RemotingException {
        if (send_reconnect && !isConnected()){
            connect();
        }
//使用com.alibaba.dubbo.remoting.transport.netty.NettyChannel傳遞資料
        Channel channel = getChannel();
        //TODO getChannel返回的狀態是否包含null需要改進
        if (channel == null || ! channel.isConnected()) {
          throw new RemotingException(this, "message can not send, because channel is closed . url:" + getUrl());
        }
        channel.send(message, sent);
    }
//最後一下
com.alibaba.dubbo.remoting.transport.netty.NettyChannel.send(Object, boolean)

public void send(Object message, boolean sent) throws RemotingException {
        super.send(message, sent);
        
        boolean success = true;
        int timeout = 0;
        try {
//呼叫netty  org.jboss.netty.channel.Channel.write(Object)
            ChannelFuture future = channel.write(message);
            if (sent) {
                timeout = getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
                //如果是非同步呼叫
success = future.await(timeout);
            }
            Throwable cause = future.getCause();
            if (cause != null) {
                throw cause;
            }
        } catch (Throwable e) {
            throw new RemotingException(this, "Failed to send message " + message + " to " + getRemoteAddress() + ", cause: " + e.getMessage(), e);
        }
        
        if(! success) {
            throw new RemotingException(this, "Failed to send message " + message + " to " + getRemoteAddress()
                    + "in timeout(" + timeout + "ms) limit");
        }
    }

相關推薦

dubbo 服務呼叫原始碼分析

下面是呼叫方法棧和核心程式碼分析InvokerInvocationHandler.invoke(Object, Method, Object[]) line: 38com.alibaba.dubbo.rpc.cluster.support.wrapper.MockCluste

dubbo遠端呼叫原始碼分析

dubbo遠端呼叫原始碼分析（一）：客戶端傳送請求 本文分為三部分，分別是： 消費端註冊部分 消費端動態代理部分 消費端事件處理器部分 https://blog.csdn.net/lkforce/article/details/79816921 ---------------

dubbo遠端呼叫原始碼分析（一）：客戶端傳送請求

dubbo遠端呼叫的原始碼分析，分成了三篇文章地址分別如下：本文分為三部分，分別是：消費端註冊部分消費端動態代理部分消費端事件處理器部分消費端註冊部分在分析dubbo遠端呼叫的時候，要從dubbo消費端（consumer）註冊開始說起dubbo的consumer在註冊的時候，

dubbo遠端呼叫原始碼分析（三）：客戶端接收反饋後的處理

dubbo遠端呼叫的原始碼分析，分成了三篇文章地址分別如下：下面是consumer接收到provider反饋時的處理consumer接收到provider的反饋後，觸發NettyClient的事件處理器，該事件對consumer來說是上行事件，觸發的是NettyCodecAd

Dubbo服務呼叫過程原始碼解析④

[TOC] > [Dubbo SPI原始碼解析①](https://www.cnblogs.com/lbhym/p/14192704.html) > > [Dubbo服務暴露原始碼解析②](https://www.cnblogs.com/lbhym/p/14192711.html) > > [Dubbo服務

Dubbo服務呼叫Failed to invoke the method錯誤記錄

Dubbo服務呼叫Failed to invoke the method錯誤記錄 在開發過程中我遇到一個問題： 一個多模組專案，服務與應用之間採用dubbo進行呼叫，啟動服務後用瀏覽器訪問一切都好，但當採用fiddler進行模擬外系統請求時卻死活調不通，報錯如下： [ERR

Dubbo的Extension原始碼分析

我們基於ExtensionLoader.getExtensionLoader().getAdaptiveExtension()這個入口進行了原始碼分析，已經通過上一節課進行了分析。我也做了很詳細的筆記給大家去做鞏固，希望大家有去學習 簡單整理一下上節課getAdaptiveExtension的流程圖

Dubbo服務呼叫原理

服務呼叫原理 引用服務 最終，建立一個代理物件 InvokerInvocationHandler Invoke，是一層一層封裝的結果 invoker.invoke 執行 MockClusterInvoker invok

使用Zipkin和Brave 實現dubbo服務呼叫跟蹤

git專案地址：https://github.com/blacklau/http-dubbo-zipkin(點選開啟連結)，請下載使用。 本工程通過模擬訂單詳情的查詢，演示系統的呼叫鏈跟蹤，跟蹤資訊包括呼叫服務及請求引數。 涉及的各工程作用: louie-webap

Dubbo——服務暴露過程分析

    這篇文章來敘述對dubbo的服務暴露前的準備工作： 使用Spring配置檔案，通過main方法來啟動spring容器，來觀察dubbo服務的啟動過程。 dubbo配置檔案 <context:component-scan base-package="

Mybatis之Mapper呼叫原始碼分析

Mybatis之Mapper呼叫原始碼分析 這一篇是承接前面兩篇的，分別為:Mybatis原始碼解析之配置載入(一), Mybatis原始碼解析之配置載入(二),前面兩篇講了在Mybatis啟動時如何載入配置，這一節就講在執行時，如何通過session獲取Mapper代理類，從而實現

dubbo服務呼叫

一：消費端傳送請求 1.當呼叫dubbo介面方法時，因為獲取的類例項是FactoryBean介面返回的代理類，所以會先經過InvokerInvocationHandler的invoke方法，這個代理是在類初始化時設定的 public Object invoke(Object

Dubbo 服務呼叫返回的物件部分屬性返回為null

 最近呼叫dubbo服務時有兩個奇怪現象；controller接收到bean引數後都有值，但是當呼叫服務時傳過去的引數就部分為null了（比如remark在controller裡檢視bean物件是有值的，但是到facade服務層傳遞過去的bean中remark就沒有值了）；還

spring 自動注入和 dubbo服務呼叫問題

今天 遇到這樣一個問題： 工程A中我寫寫了一個spring security的自定義認證類，這個類始終不能由容器管理，但是這個時候 我需要通過spring 自動注入某個服務類來獲取使用者相關資訊 來進行認證） 問題出現在： 自定義認證類不是由容器管

dubbo叢集容錯原始碼分析

首先我們看一下官網的文件：啟動專案後，為了走到負載均衡，我們需要啟動不少於2個的提供者大致呼叫棧如下：sayHello:-1, proxy0 (com.alibaba.dubbo.common.bytecode)invoke:51, InvokerInvocationHand

阿里巴巴Dubbo實現的原始碼分析

1.     Dubbo概述 Dubbo是阿里巴巴開源出來的一個分散式服務框架，致力於提供高效能和透明化的RPC遠端服務呼叫方案，以及作為SOA服務治理的方案。它的核心功能包括： ＃remoting: 遠端通訊基礎，提供對多種NIO框架抽象封裝，包括“同步轉非同步”和“請求

Dubbo 服務呼叫原理淺析

dubbo概念dubbo原理dubbo應用場景    Dubbo概念：      Dubbo是一個分散式服務框架，致力於提供高效能和透明化的RPC遠端服務呼叫方案，以及SOA服務治理方案。簡單的說，du

Dubbo服務呼叫，時好時壞，一會兒呼叫正常，一會兒呼叫不正常？

摘要：昨天上線個新版本的dubbo服務到預釋出環境，今天來驗證功能，發現，其中一個介面呼叫，1次正常，下一次就不正常，再重新整理又正常了，這到底是什麼問題呢，我們來分析下： 一：問題排查： 1.首先檢視是不是部署了多臺服務提供者，如果是停掉其中一臺，這麼做以後，發現還是有

Dubbo服務暴露原始碼解析②

[TOC] ​ 先放一張官網的服務暴露時序圖，對我們梳理原始碼有很大的幫助。注：不論是暴露還是匯出或者是其他翻譯，都是描述export的，只是翻譯不同。 ![](https://img2020.cnblogs.com/blog/1383122/202012/1383122-2020122613493559

Dubbo SPI 機制原始碼分析（基於2.7.7）

Dubbo SPI 機制涉及到 `@SPI`、`@Adaptive`、`@Activate` 三個註解，ExtensionLoader 作為 Dubbo SPI 機制的核心負責載入和管理擴充套件點及其實現。本文以 ExtensionLoader 的原始碼作為分析主線，進而引出三個註解的作用和工作機制。 Ex