摘要：遠(yuǎn)程調(diào)用開篇目標(biāo)介紹之后解讀遠(yuǎn)程調(diào)用模塊的內(nèi)容如何編排介紹中的包結(jié)構(gòu)設(shè)計(jì)以及最外層的的源碼解析。十該類就是遠(yuǎn)程調(diào)用的上下文，貫穿著整個(gè)調(diào)用，例如調(diào)用，然后調(diào)用。十五該類是系統(tǒng)上下文，僅供內(nèi)部使用。

目標(biāo)：介紹之后解讀遠(yuǎn)程調(diào)用模塊的內(nèi)容如何編排、介紹dubbo-rpc-api中的包結(jié)構(gòu)設(shè)計(jì)以及最外層的的源碼解析。

最近我面臨著一個(gè)選擇，因?yàn)閐ubbo 2.7.0-release出現(xiàn)在了倉庫里，最近一直在進(jìn)行2.7.0版本的code review，那我之前說這一系列的文章都是講述2.6.x版本的源代碼，我現(xiàn)在要不要選擇直接開始講解2.7.0的版本的源碼呢？我最后還是決定繼續(xù)講解2.6.x，因?yàn)槲矣X得還是有很多公司在用著2.6.x的版本，并且對(duì)于升級(jí)2.7.0的計(jì)劃應(yīng)該還沒那么快，并且在了解2.6.x版本的原理后，再去了解2.7.0新增的特性會(huì)更加容易，也能夠品位到設(shè)計(jì)者的意圖。當(dāng)然在結(jié)束2.6.x的重要模塊講解后，我也會(huì)對(duì)2.7.0的新特性以及實(shí)現(xiàn)原理做一個(gè)全面的分析，2.7.0作為dubbo社區(qū)的畢業(yè)版，更加強(qiáng)大，敬請(qǐng)期待。

前面講了很多的內(nèi)容，現(xiàn)在開始將遠(yuǎn)程調(diào)用RPC，好像又回到我第一篇文章 《dubbo源碼解析（一）Hello,Dubbo》，在這篇文章開頭我講到了什么叫做RPC，再通俗一點(diǎn)講，就是我把一個(gè)項(xiàng)目的兩部分代碼分開來，分別放到兩臺(tái)機(jī)器上，當(dāng)我部署在A服務(wù)器上的應(yīng)用想要調(diào)用部署在B服務(wù)器上的應(yīng)用等方法，由于不存在同一個(gè)內(nèi)存空間，不能直接調(diào)用。而其實(shí)整個(gè)dubbo都在做遠(yuǎn)程調(diào)用的事情，它涉及到很多內(nèi)容，比如配置、代理、集群、監(jiān)控等等，那么這次講的內(nèi)容是只關(guān)心一對(duì)一的調(diào)用，dubbo-rpc遠(yuǎn)程調(diào)用模塊抽象各種協(xié)議，以及動(dòng)態(tài)代理，Proxy層和Protocol層rpc的核心，我將會(huì)在本系列中講到。下面我們來看兩張官方文檔的圖：

暴露服務(wù)的時(shí)序圖：

你會(huì)發(fā)現(xiàn)其中有我們以前講到的Transporter、Server、Registry，而這次的系列將會(huì)講到的就是紅色框框內(nèi)的部分。

引用服務(wù)時(shí)序圖

在引用服務(wù)時(shí)序圖中，對(duì)應(yīng)的也是紅色框框的部分。

當(dāng)閱讀完該系列后，希望能對(duì)這個(gè)調(diào)用鏈有所感悟。接下來看看dubbo-rpc的包結(jié)構(gòu)：

可以看到有很多包，很規(guī)整，其中dubbo-rpc-api是對(duì)協(xié)議、暴露、引用、代理等的抽象和實(shí)現(xiàn)，是rpc整個(gè)設(shè)計(jì)的核心內(nèi)容。其他的包則是dubbo支持的9種協(xié)議，在官方文檔也能查看介紹，并且包括一種本地調(diào)用injvm。那么我們?cè)賮砜纯磀ubbo-rpc-api中包結(jié)構(gòu)：

filter包：在進(jìn)行服務(wù)引用時(shí)會(huì)進(jìn)行一系列的過濾。其中包括了很多過濾器。

listener包：看上面兩張服務(wù)引用和服務(wù)暴露的時(shí)序圖，發(fā)現(xiàn)有兩個(gè)listener，其中的邏輯實(shí)現(xiàn)就在這個(gè)包內(nèi)

protocol包：這個(gè)包實(shí)現(xiàn)了協(xié)議的一些公共邏輯

proxy包：實(shí)現(xiàn)了代理的邏輯。

service包：其中包含了一個(gè)需要調(diào)用的方法等封裝抽象。

support包：包括了工具類

最外層的實(shí)現(xiàn)。

下面的篇幅設(shè)計(jì)，本文會(huì)講解最外層的源碼和service下的源碼，support包下的源碼我會(huì)穿插在其他用到的地方一并講解，filter、listener、protocol、proxy以及各類協(xié)議的實(shí)現(xiàn)各自用一篇來講。

public interface Invoker extends Node {

    /**
     * get service interface.
     * 獲得服務(wù)接口
     * @return service interface.
     */
    Class getInterface();

    /**
     * invoke.
     * 調(diào)用下一個(gè)會(huì)話域
     * @param invocation
     * @return result
     * @throws RpcException
     */
    Result invoke(Invocation invocation) throws RpcException;

}

該接口是實(shí)體域，它是dubbo的核心模型，其他模型都向它靠攏，或者轉(zhuǎn)化成它，它代表了一個(gè)可執(zhí)行體，可以向它發(fā)起invoke調(diào)用，這個(gè)有可能是一個(gè)本地的實(shí)現(xiàn)，也可能是一個(gè)遠(yuǎn)程的實(shí)現(xiàn)，也可能是一個(gè)集群的實(shí)現(xiàn)。它代表了一次調(diào)用

public interface Invocation {

    /**
     * get method name.
     * 獲得方法名稱
     * @return method name.
     * @serial
     */
    String getMethodName();

    /**
     * get parameter types.
     * 獲得參數(shù)類型
     * @return parameter types.
     * @serial
     */
    Class[] getParameterTypes();

    /**
     * get arguments.
     * 獲得參數(shù)
     * @return arguments.
     * @serial
     */
    Object[] getArguments();

    /**
     * get attachments.
     * 獲得附加值集合
     * @return attachments.
     * @serial
     */
    Map getAttachments();

    /**
     * get attachment by key.
     * 獲得附加值
     * @return attachment value.
     * @serial
     */
    String getAttachment(String key);

    /**
     * get attachment by key with default value.
     * 獲得附加值
     * @return attachment value.
     * @serial
     */
    String getAttachment(String key, String defaultValue);

    /**
     * get the invoker in current context.
     * 獲得當(dāng)前上下文的invoker
     * @return invoker.
     * @transient
     */
    Invoker getInvoker();

}

Invocation 是會(huì)話域，它持有調(diào)用過程中的變量，比如方法名，參數(shù)等。

public interface Exporter {

    /**
     * get invoker.
     * 獲得對(duì)應(yīng)的實(shí)體域invoker
     * @return invoker
     */
    Invoker getInvoker();

    /**
     * unexport.
     * 取消暴露
     * 

     * 
     * getInvoker().destroy();
     * 
     */
    void unexport();

}

該接口是暴露服務(wù)的接口，定義了兩個(gè)方法分別是獲得invoker和取消暴露服務(wù)。

@SPI
public interface ExporterListener {

    /**
     * The exporter exported.
     * 暴露服務(wù)
     * @param exporter
     * @throws RpcException
     * @see com.alibaba.dubbo.rpc.Protocol#export(Invoker)
     */
    void exported(Exporter exporter) throws RpcException;

    /**
     * The exporter unexported.
     * 取消暴露
     * @param exporter
     * @throws RpcException
     * @see com.alibaba.dubbo.rpc.Exporter#unexport()
     */
    void unexported(Exporter exporter);

}

該接口是服務(wù)暴露的監(jiān)聽器接口，定義了兩個(gè)方法是暴露和取消暴露，參數(shù)都是Exporter類型的。

@SPI("dubbo")
public interface Protocol {

    /**
     * Get default port when user doesn"t config the port.
     * 獲得默認(rèn)的端口
     * @return default port
     */
    int getDefaultPort();

    /**
     * Export service for remote invocation: 

     * 1. Protocol should record request source address after receive a request:
     * RpcContext.getContext().setRemoteAddress();

     * 2. export() must be idempotent, that is, there"s no difference between invoking once and invoking twice when
     * export the same URL

     * 3. Invoker instance is passed in by the framework, protocol needs not to care 

     * 暴露服務(wù)方法，
     * @param      Service type 服務(wù)類型
     * @param invoker Service invoker 服務(wù)的實(shí)體域
     * @return exporter reference for exported service, useful for unexport the service later
     * @throws RpcException thrown when error occurs during export the service, for example: port is occupied
     */
    @Adaptive
     Exporter export(Invoker invoker) throws RpcException;

    /**
     * Refer a remote service: 

     * 1. When user calls `invoke()` method of `Invoker` object which"s returned from `refer()` call, the protocol
     * needs to correspondingly execute `invoke()` method of `Invoker` object 

     * 2. It"s protocol"s responsibility to implement `Invoker` which"s returned from `refer()`. Generally speaking,
     * protocol sends remote request in the `Invoker` implementation. 

     * 3. When there"s check=false set in URL, the implementation must not throw exception but try to recover when
     * connection fails.
     * 引用服務(wù)方法
     * @param   Service type 服務(wù)類型
     * @param type Service class 服務(wù)類名
     * @param url  URL address for the remote service
     * @return invoker service"s local proxy
     * @throws RpcException when there"s any error while connecting to the service provider
     */
    @Adaptive
     Invoker refer(Class type, URL url) throws RpcException;

    /**
     * Destroy protocol: 

     * 1. Cancel all services this protocol exports and refers 

     * 2. Release all occupied resources, for example: connection, port, etc. 

     * 3. Protocol can continue to export and refer new service even after it"s destroyed.
     */
    void destroy();

}

該接口是服務(wù)域接口，也是協(xié)議接口，它是一個(gè)可擴(kuò)展的接口，默認(rèn)實(shí)現(xiàn)的是dubbo協(xié)議。定義了四個(gè)方法，關(guān)鍵的是服務(wù)暴露和引用兩個(gè)方法。

@SPI
public interface Filter {

    /**
     * do invoke filter.
     * 

     * 
     * // before filter
     * Result result = invoker.invoke(invocation);
     * // after filter
     * return result;
     * 
     *
     * @param invoker    service
     * @param invocation invocation.
     * @return invoke result.
     * @throws RpcException
     * @see com.alibaba.dubbo.rpc.Invoker#invoke(Invocation)
     */
    Result invoke(Invoker invoker, Invocation invocation) throws RpcException;

}

該接口是invoker調(diào)用時(shí)過濾器接口，其中就只有一個(gè)invoke方法。在該方法中對(duì)調(diào)用進(jìn)行過濾

@SPI
public interface InvokerListener {

    /**
     * The invoker referred
     * 在服務(wù)引用的時(shí)候進(jìn)行監(jiān)聽
     * @param invoker
     * @throws RpcException
     * @see com.alibaba.dubbo.rpc.Protocol#refer(Class, com.alibaba.dubbo.common.URL)
     */
    void referred(Invoker invoker) throws RpcException;

    /**
     * The invoker destroyed.
     * 銷毀實(shí)體域
     * @param invoker
     * @see com.alibaba.dubbo.rpc.Invoker#destroy()
     */
    void destroyed(Invoker invoker);

}

該接口是實(shí)體域的監(jiān)聽器，定義了兩個(gè)方法，分別是服務(wù)引用和銷毀的時(shí)候執(zhí)行的方法。

該接口是實(shí)體域執(zhí)行invoke的結(jié)果接口，里面定義了獲得結(jié)果異常以及附加值等方法。比較好理解我就不貼代碼了。

@SPI("javassist")
public interface ProxyFactory {

    /**
     * create proxy.
     * 創(chuàng)建一個(gè)代理
     * @param invoker
     * @return proxy
     */
    @Adaptive({Constants.PROXY_KEY})
     T getProxy(Invoker invoker) throws RpcException;

    /**
     * create proxy.
     * 創(chuàng)建一個(gè)代理
     * @param invoker
     * @return proxy
     */
    @Adaptive({Constants.PROXY_KEY})
     T getProxy(Invoker invoker, boolean generic) throws RpcException;

    /**
     * create invoker.
     * 創(chuàng)建一個(gè)實(shí)體域
     * @param 
     * @param proxy
     * @param type
     * @param url
     * @return invoker
     */
    @Adaptive({Constants.PROXY_KEY})
     Invoker getInvoker(T proxy, Class type, URL url) throws RpcException;

}

該接口是代理工廠接口，它也是個(gè)可擴(kuò)展接口，默認(rèn)實(shí)現(xiàn)javassist，dubbo提供兩種動(dòng)態(tài)代理方法分別是javassist/jdk，該接口定義了三個(gè)方法，前兩個(gè)方法是通過invoker創(chuàng)建代理，最后一個(gè)是通過代理來獲得invoker。

該類就是遠(yuǎn)程調(diào)用的上下文，貫穿著整個(gè)調(diào)用，例如A調(diào)用B，然后B調(diào)用C。在服務(wù)B上，RpcContext在B之前將調(diào)用信息從A保存到B。開始調(diào)用C，并在B調(diào)用C后將調(diào)用信息從B保存到C。RpcContext保存了調(diào)用信息。

public class RpcContext {

    /**
     * use internal thread local to improve performance
     * 本地上下文
     */
    private static final InternalThreadLocal LOCAL = new InternalThreadLocal() {
        @Override
        protected RpcContext initialValue() {
            return new RpcContext();
        }
    };
    /**
     * 服務(wù)上下文
     */
    private static final InternalThreadLocal SERVER_LOCAL = new InternalThreadLocal() {
        @Override
        protected RpcContext initialValue() {
            return new RpcContext();
        }
    };

    /**
     * 附加值集合
     */
    private final Map attachments = new HashMap();
    /**
     * 上下文值
     */
    private final Map values = new HashMap();
    /**
     * 線程結(jié)果
     */
    private Future future;

    /**
     * url集合
     */
    private List urls;

    /**
     * 當(dāng)前的url
     */
    private URL url;

    /**
     * 方法名稱
     */
    private String methodName;

    /**
     * 參數(shù)類型集合
     */
    private Class[] parameterTypes;

    /**
     * 參數(shù)集合
     */
    private Object[] arguments;

    /**
     * 本地地址
     */
    private InetSocketAddress localAddress;

    /**
     * 遠(yuǎn)程地址
     */
    private InetSocketAddress remoteAddress;
    /**
     * 實(shí)體域集合
     */
    @Deprecated
    private List> invokers;
    /**
     * 實(shí)體域
     */
    @Deprecated
    private Invoker invoker;
    /**
     * 會(huì)話域
     */
    @Deprecated
    private Invocation invocation;

    // now we don"t use the "values" map to hold these objects
    // we want these objects to be as generic as possible
    /**
     * 請(qǐng)求
     */
    private Object request;
    /**
     * 響應(yīng)
     */
    private Object response;

該類中最重要的是它的一些屬性，因?yàn)樵撋舷挛木褪怯脕肀４嫘畔⒌?。方法我就不介紹了，因?yàn)楸容^簡單。

/**
 * 不知道異常
 */
public static final int UNKNOWN_EXCEPTION = 0;
/**
 * 網(wǎng)絡(luò)異常
 */
public static final int NETWORK_EXCEPTION = 1;
/**
 * 超時(shí)異常
 */
public static final int TIMEOUT_EXCEPTION = 2;
/**
 * 基礎(chǔ)異常
 */
public static final int BIZ_EXCEPTION = 3;
/**
 * 禁止訪問異常
 */
public static final int FORBIDDEN_EXCEPTION = 4;
/**
 * 序列化異常
 */
public static final int SERIALIZATION_EXCEPTION = 5;

該類是rpc調(diào)用拋出的異常類，其中封裝了五種通用的錯(cuò)誤碼。

/**
 * 方法名稱
 */
private String methodName;

/**
 * 參數(shù)類型集合
 */
private Class[] parameterTypes;

/**
 * 參數(shù)集合
 */
private Object[] arguments;

/**
 * 附加值
 */
private Map attachments;

/**
 * 實(shí)體域
 */
private transient Invoker invoker;

該類實(shí)現(xiàn)了Invocation接口，是rpc的會(huì)話域，其中的方法比較簡單，主要是封裝了上述的屬性。

/**
 * 結(jié)果
 */
private Object result;

/**
 * 異常
 */
private Throwable exception;

/**
 * 附加值
 */
private Map attachments = new HashMap();

該類實(shí)現(xiàn)了Result接口，是rpc的結(jié)果實(shí)現(xiàn)類，其中關(guān)鍵是封裝了以上三個(gè)屬性。

該類是rpc的一些狀態(tài)監(jiān)控，其中封裝了許多的計(jì)數(shù)器，用來記錄rpc調(diào)用的狀態(tài)。

/**
 * uri對(duì)應(yīng)的狀態(tài)集合，key為uri，value為RpcStatus對(duì)象
 */
private static final ConcurrentMap SERVICE_STATISTICS = new ConcurrentHashMap();

/**
 * method對(duì)應(yīng)的狀態(tài)集合，key是uri，第二個(gè)key是方法名methodName
 */
private static final ConcurrentMap> METHOD_STATISTICS = new ConcurrentHashMap>();
/**
 * 已經(jīng)沒用了
 */
private final ConcurrentMap values = new ConcurrentHashMap();
/**
 * 活躍狀態(tài)
 */
private final AtomicInteger active = new AtomicInteger();
/**
 * 總的數(shù)量
 */
private final AtomicLong total = new AtomicLong();
/**
 * 失敗的個(gè)數(shù)
 */
private final AtomicInteger failed = new AtomicInteger();
/**
 * 總調(diào)用時(shí)長
 */
private final AtomicLong totalElapsed = new AtomicLong();
/**
 * 總調(diào)用失敗時(shí)長
 */
private final AtomicLong failedElapsed = new AtomicLong();
/**
 * 最大調(diào)用時(shí)長
 */
private final AtomicLong maxElapsed = new AtomicLong();
/**
 * 最大調(diào)用失敗時(shí)長
 */
private final AtomicLong failedMaxElapsed = new AtomicLong();
/**
 * 最大調(diào)用成功時(shí)長
 */
private final AtomicLong succeededMaxElapsed = new AtomicLong();

/**
 * Semaphore used to control concurrency limit set by `executes`
 * 信號(hào)量用來控制`execution`設(shè)置的并發(fā)限制
 */
private volatile Semaphore executesLimit;
/**
 * 用來控制`execution`設(shè)置的許可證
 */
private volatile int executesPermits;

以上是該類的屬性，可以看到保存了很多的計(jì)數(shù)器，分別用來記錄了失敗調(diào)用成功調(diào)用等累計(jì)數(shù)。

/**
 * 開始計(jì)數(shù)
 * @param url
 */
public static void beginCount(URL url, String methodName) {
    // 對(duì)該url對(duì)應(yīng)對(duì)活躍計(jì)數(shù)器加一
    beginCount(getStatus(url));
    // 對(duì)該方法對(duì)活躍計(jì)數(shù)器加一
    beginCount(getStatus(url, methodName));
}

/**
 * 以原子方式加1
 * @param status
 */
private static void beginCount(RpcStatus status) {
    status.active.incrementAndGet();
}

該方法是增加計(jì)數(shù)。

public static void endCount(URL url, String methodName, long elapsed, boolean succeeded) {
    // url對(duì)應(yīng)的狀態(tài)中計(jì)數(shù)器減一
    endCount(getStatus(url), elapsed, succeeded);
    // 方法對(duì)應(yīng)的狀態(tài)中計(jì)數(shù)器減一
    endCount(getStatus(url, methodName), elapsed, succeeded);
}

private static void endCount(RpcStatus status, long elapsed, boolean succeeded) {
    // 活躍計(jì)數(shù)器減一
    status.active.decrementAndGet();
    // 總計(jì)數(shù)器加1
    status.total.incrementAndGet();
    // 總調(diào)用時(shí)長加上調(diào)用時(shí)長
    status.totalElapsed.addAndGet(elapsed);
    // 如果最大調(diào)用時(shí)長小于elapsed，則設(shè)置最大調(diào)用時(shí)長
    if (status.maxElapsed.get() < elapsed) {
        status.maxElapsed.set(elapsed);
    }
    // 如果rpc調(diào)用成功
    if (succeeded) {
        // 如果成最大調(diào)用成功時(shí)長小于elapsed，則設(shè)置最大調(diào)用成功時(shí)長
        if (status.succeededMaxElapsed.get() < elapsed) {
            status.succeededMaxElapsed.set(elapsed);
        }
    } else {
        // 失敗計(jì)數(shù)器加一
        status.failed.incrementAndGet();
        // 失敗的過期數(shù)加上elapsed
        status.failedElapsed.addAndGet(elapsed);
        // 總調(diào)用失敗時(shí)長小于elapsed，則設(shè)置總調(diào)用失敗時(shí)長
        if (status.failedMaxElapsed.get() < elapsed) {
            status.failedMaxElapsed.set(elapsed);
        }
    }
}

該方法是計(jì)數(shù)器減少。

該類是系統(tǒng)上下文，僅供內(nèi)部使用。

/**
 * 系統(tǒng)名稱
 */
private static final String SYSTEMNAME = "system";
/**
 * 系統(tǒng)上下文集合，僅供內(nèi)部使用
 */
private static final ConcurrentMap context_map = new ConcurrentHashMap();
/**
 * 系統(tǒng)上下文名稱
 */
private String name;

上面是該類的屬性，它還記錄了所有的系統(tǒng)上下文集合。

public interface EchoService {

    /**
     * echo test.
     * 回聲測試
     * @param message message.
     * @return message.
     */
    Object $echo(Object message);

}

該接口是回聲服務(wù)接口，定義了一個(gè)一個(gè)回聲測試的方法，回聲測試用于檢測服務(wù)是否可用，回聲測試按照正常請(qǐng)求流程執(zhí)行，能夠測試整個(gè)調(diào)用是否通暢，可用于監(jiān)控，所有服務(wù)自動(dòng)實(shí)現(xiàn)該接口，只需將任意服務(wù)強(qiáng)制轉(zhuǎn)化為EchoService，就可以用了。

該方法是通用的異常類。

/**
 * 異常類名
 */
private String exceptionClass;

/**
 * 異常信息
 */
private String exceptionMessage;

比較簡單，就封裝了兩個(gè)屬性。

public interface GenericService {

    /**
     * Generic invocation
     * 通用的會(huì)話域
     * @param method         Method name, e.g. findPerson. If there are overridden methods, parameter info is
     *                       required, e.g. findPerson(java.lang.String)
     * @param parameterTypes Parameter types
     * @param args           Arguments
     * @return invocation return value
     * @throws Throwable potential exception thrown from the invocation
     */
    Object $invoke(String method, String[] parameterTypes, Object[] args) throws GenericException;

}

該接口是通用的服務(wù)接口，同樣定義了一個(gè)類似invoke的方法

該部分相關(guān)的源碼解析地址：https://github.com/CrazyHZM/i...

該文章講解了遠(yuǎn)程調(diào)用的開篇，介紹之后解讀遠(yuǎn)程調(diào)用模塊的內(nèi)容如何編排、介紹dubbo-rpc-api中的包結(jié)構(gòu)設(shè)計(jì)以及最外層的的源碼解析，其中的邏輯不負(fù)責(zé)，要關(guān)注的是其中的一些概念和dubbo如何去做暴露服務(wù)和引用服務(wù)，其中很多的接口定義需要弄清楚。接下來我將開始對(duì)rpc模塊的過濾器進(jìn)行講解。