首頁(yè)技術(shù)文章正文

Java培訓(xùn):dubbo源碼解析-網(wǎng)絡(luò)通信

更新時(shí)間:2022-09-16 來(lái)源:黑馬程序員 瀏覽量:

  在之前的內(nèi)容中,我們講解了消費(fèi)者端服務(wù)發(fā)現(xiàn)與提供者端服務(wù)暴露的相關(guān)內(nèi)容,同時(shí)也知道消費(fèi)者端通過(guò)內(nèi)置的負(fù)載均衡算法獲取合適的調(diào)用invoker進(jìn)行遠(yuǎn)程調(diào)用。那么,本章節(jié)重點(diǎn)關(guān)注的就是遠(yuǎn)程調(diào)用過(guò)程即網(wǎng)絡(luò)通信。

  

1663308753668_1.jpg

  網(wǎng)絡(luò)通信位于Remoting模塊:

  - Remoting 實(shí)現(xiàn)是 Dubbo 協(xié)議的實(shí)現(xiàn),如果你選擇 RMI 協(xié)議,整個(gè) Remoting 都不會(huì)用上;

  - Remoting 內(nèi)部再劃為 `Transport 傳輸層` 和 `Exchange 信息交換層`;

  - Transport 層只負(fù)責(zé)單向消息傳輸,是對(duì) Mina, Netty, Grizzly 的抽象,它也可以擴(kuò)展 UDP 傳輸;

  - Exchange 層是在傳輸層之上封裝了 Request-Response 語(yǔ)義;

  網(wǎng)絡(luò)通信的問(wèn)題:

  客戶端與服務(wù)端連通性問(wèn)題

  粘包拆包問(wèn)題

  異步多線程數(shù)據(jù)一致問(wèn)題

  通信協(xié)議

  dubbo內(nèi)置,dubbo協(xié)議 ,rmi協(xié)議,hessian協(xié)議,http協(xié)議,webservice協(xié)議,thrift協(xié)議,rest協(xié)議,grpc協(xié)議,memcached協(xié)議,redis協(xié)議等10種通訊協(xié)議。各個(gè)協(xié)議特點(diǎn)如下

  dubbo協(xié)議

  Dubbo 缺省協(xié)議采用單一長(zhǎng)連接和 NIO 異步通訊,適合于小數(shù)據(jù)量大并發(fā)的服務(wù)調(diào)用,以及服務(wù)消費(fèi)者機(jī)器數(shù)遠(yuǎn)大于服務(wù)提供者機(jī)器數(shù)的情況。

  缺省協(xié)議,使用基于 mina `1.1.7` 和 hessian `3.2.1` 的 tbremoting 交互。

  - 連接個(gè)數(shù):?jiǎn)芜B接

  - 連接方式:長(zhǎng)連接

  - 傳輸協(xié)議:TCP

  - 傳輸方式:NIO 異步傳輸

  - 序列化:Hessian 二進(jìn)制序列化

  - 適用范圍:傳入傳出參數(shù)數(shù)據(jù)包較小(建議小于100K),消費(fèi)者比提供者個(gè)數(shù)多,單一消費(fèi)者無(wú)法壓滿提供者,盡量不要用 dubbo 協(xié)議傳輸大文件或超大字符串。

  - 適用場(chǎng)景:常規(guī)遠(yuǎn)程服務(wù)方法調(diào)用

  rmi協(xié)議

  RMI 協(xié)議采用 JDK 標(biāo)準(zhǔn)的 `java.rmi.*` 實(shí)現(xiàn),采用阻塞式短連接和 JDK 標(biāo)準(zhǔn)序列化方式。

  - 連接個(gè)數(shù):多連接

  - 連接方式:短連接

  - 傳輸協(xié)議:TCP

  - 傳輸方式:同步傳輸

  - 序列化:Java 標(biāo)準(zhǔn)二進(jìn)制序列化

  - 適用范圍:傳入傳出參數(shù)數(shù)據(jù)包大小混合,消費(fèi)者與提供者個(gè)數(shù)差不多,可傳文件。

  - 適用場(chǎng)景:常規(guī)遠(yuǎn)程服務(wù)方法調(diào)用,與原生RMI服務(wù)互操作

  hessian協(xié)議

  Hessian 協(xié)議用于集成 Hessian 的服務(wù),Hessian 底層采用 Http 通訊,采用 Servlet 暴露服務(wù),Dubbo 缺省內(nèi)嵌 Jetty 作為服務(wù)器實(shí)現(xiàn)。

  Dubbo 的 Hessian 協(xié)議可以和原生 Hessian 服務(wù)互操作,即:

  - 提供者用 Dubbo 的 Hessian 協(xié)議暴露服務(wù),消費(fèi)者直接用標(biāo)準(zhǔn) Hessian 接口調(diào)用

  - 或者提供方用標(biāo)準(zhǔn) Hessian 暴露服務(wù),消費(fèi)方用 Dubbo 的 Hessian 協(xié)議調(diào)用。

  - 連接個(gè)數(shù):多連接

  - 連接方式:短連接

  - 傳輸協(xié)議:HTTP

  - 傳輸方式:同步傳輸

  - 序列化:Hessian二進(jìn)制序列化

  - 適用范圍:傳入傳出參數(shù)數(shù)據(jù)包較大,提供者比消費(fèi)者個(gè)數(shù)多,提供者壓力較大,可傳文件。

  - 適用場(chǎng)景:頁(yè)面?zhèn)鬏?,文件傳輸,或與原生hessian服務(wù)互操作

  http協(xié)議

  基于 HTTP 表單的遠(yuǎn)程調(diào)用協(xié)議,采用 Spring 的 HttpInvoker 實(shí)現(xiàn)

  - 連接個(gè)數(shù):多連接

  - 連接方式:短連接

  - 傳輸協(xié)議:HTTP

  - 傳輸方式:同步傳輸

  - 序列化:表單序列化

  - 適用范圍:傳入傳出參數(shù)數(shù)據(jù)包大小混合,提供者比消費(fèi)者個(gè)數(shù)多,可用瀏覽器查看,可用表單或URL傳入?yún)?shù),暫不支持傳文件。

  - 適用場(chǎng)景:需同時(shí)給應(yīng)用程序和瀏覽器 JS 使用的服務(wù)。

  webservice協(xié)議

  基于 WebService 的遠(yuǎn)程調(diào)用協(xié)議,基于 Apache CXF 實(shí)現(xiàn)](http://dubbo.apache.org/zh-cn/docs/user/references/protocol/webservice.html#fn2)。

  可以和原生 WebService 服務(wù)互操作,即:

  - 提供者用 Dubbo 的 WebService 協(xié)議暴露服務(wù),消費(fèi)者直接用標(biāo)準(zhǔn) WebService 接口調(diào)用,

  - 或者提供方用標(biāo)準(zhǔn) WebService 暴露服務(wù),消費(fèi)方用 Dubbo 的 WebService 協(xié)議調(diào)用。

  - 連接個(gè)數(shù):多連接

  - 連接方式:短連接

  - 傳輸協(xié)議:HTTP

  - 傳輸方式:同步傳輸

  - 序列化:SOAP 文本序列化(http + xml)

  - 適用場(chǎng)景:系統(tǒng)集成,跨語(yǔ)言調(diào)用

  thrift協(xié)議

  當(dāng)前 dubbo 支持 [[1\]](http://dubbo.apache.org/zh-cn/docs/user/references/protocol/thrift.html#fn1)的 thrift 協(xié)議是對(duì) thrift 原生協(xié)議 [[2\]](http://dubbo.apache.org/zh-cn/docs/user/references/protocol/thrift.html#fn2) 的擴(kuò)展,在原生協(xié)議的基礎(chǔ)上添加了一些額外的頭信息,比如 service name,magic number 等。

  rest協(xié)議

  基于標(biāo)準(zhǔn)的Java REST API——JAX-RS 2.0(Java API for RESTful Web Services的簡(jiǎn)寫(xiě))實(shí)現(xiàn)的REST調(diào)用支持

  grpc協(xié)議

  Dubbo 自 2.7.5 版本開(kāi)始支持 gRPC 協(xié)議,對(duì)于計(jì)劃使用 HTTP/2 通信,或者想利用 gRPC 帶來(lái)的 Stream、反壓、Reactive 編程等能力的開(kāi)發(fā)者來(lái)說(shuō), 都可以考慮啟用 gRPC 協(xié)議。

  - 為期望使用 gRPC 協(xié)議的用戶帶來(lái)服務(wù)治理能力,方便接入 Dubbo 體系

  - 用戶可以使用 Dubbo 風(fēng)格的,基于接口的編程風(fēng)格來(lái)定義和使用遠(yuǎn)程服務(wù)

  memcached協(xié)議

  基于 memcached實(shí)現(xiàn)的 RPC 協(xié)議

  redis協(xié)議

  基于 Redis 實(shí)現(xiàn)的 RPC 協(xié)議

  序列化

  序列化就是將對(duì)象轉(zhuǎn)成字節(jié)流,用于網(wǎng)絡(luò)傳輸,以及將字節(jié)流轉(zhuǎn)為對(duì)象,用于在收到字節(jié)流數(shù)據(jù)后還原成對(duì)象。序列化的優(yōu)勢(shì)有很多,例如安全性更好、可跨平臺(tái)等。我們知道dubbo基于netty進(jìn)行網(wǎng)絡(luò)通訊,在`NettyClient.doOpen()`方法中可以看到Netty的相關(guān)類

bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
    public ChannelPipeline getPipeline() {
        NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);
        ChannelPipeline pipeline = Channels.pipeline();
        pipeline.addLast("decoder", adapter.getDecoder());
        pipeline.addLast("encoder", adapter.getEncoder());
        pipeline.addLast("handler", nettyHandler);
        return pipeline;
    }
});

  然后去看NettyCodecAdapter 類最后進(jìn)入ExchangeCodec類的encodeRequest方法,如下:

protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException {
        Serialization serialization = getSerialization(channel);
        // header.
        byte[] header = new byte[HEADER_LENGTH];

  是的,就是Serialization接口,默認(rèn)是Hessian2Serialization序列化接口。

1663309013130_2.jpg

  Dubbo序列化支持java、compactedjava、nativejava、fastjson、dubbo、fst、hessian2、kryo,protostuff其中默認(rèn)hessian2。其中java、compactedjava、nativejava屬于原生java的序列化。

  - dubbo序列化:阿里尚未開(kāi)發(fā)成熟的高效java序列化實(shí)現(xiàn),阿里不建議在生產(chǎn)環(huán)境使用它。

  - **hessian2序列化:hessian是一種跨語(yǔ)言的高效二進(jìn)制序列化方式。但這里實(shí)際不是原生的hessian2序列化,而是阿里修改過(guò)的,它是dubbo RPC默認(rèn)啟用的序列化方式。**

  - json序列化:目前有兩種實(shí)現(xiàn),一種是采用的阿里的fastjson庫(kù),另一種是采用dubbo中自己實(shí)現(xiàn)的簡(jiǎn)單json庫(kù),但其實(shí)現(xiàn)都不是特別成熟,而且json這種文本序列化性能一般不如上面兩種二進(jìn)制序列化。

  - java序列化:主要是采用JDK自帶的Java序列化實(shí)現(xiàn),性能很不理想。

  最近幾年,各種新的高效序列化方式層出不窮,不斷刷新序列化性能的上限,最典型的包括:

  - 專門(mén)針對(duì)Java語(yǔ)言的:Kryo,F(xiàn)ST等等

  - 跨語(yǔ)言的:Protostuff,ProtoBuf,Thrift,Avro,MsgPack等等

  這些序列化方式的性能多數(shù)都顯著優(yōu)于 hessian2 (甚至包括尚未成熟的dubbo序列化)。所以我們可以為 dubbo 引入 Kryo 和 FST 這兩種高效 Java 來(lái)優(yōu)化 dubbo 的序列化。

  使用Kryo和FST非常簡(jiǎn)單,只需要在dubbo RPC的XML配置中添加一個(gè)屬性即可:

<dubbo:protocol name="dubbo" serialization="kryo"/>

  網(wǎng)絡(luò)通信

  dubbo中數(shù)據(jù)格式

  解決socket中數(shù)據(jù)粘包拆包問(wèn)題,一般有三種方式

  * 定長(zhǎng)協(xié)議(數(shù)據(jù)包長(zhǎng)度一致)

  * 定長(zhǎng)的協(xié)議是指協(xié)議內(nèi)容的長(zhǎng)度是固定的,比如協(xié)議byte長(zhǎng)度是50,當(dāng)從網(wǎng)絡(luò)上讀取50個(gè)byte后,就進(jìn)行decode解碼操作。定長(zhǎng)協(xié)議在讀取或者寫(xiě)入時(shí),效率比較高,因?yàn)閿?shù)據(jù)緩存的大小基本都確定了,就好比數(shù)組一樣,缺陷就是適應(yīng)性不足,以RPC場(chǎng)景為例,很難估計(jì)出定長(zhǎng)的長(zhǎng)度是多少。

  * 特殊結(jié)束符(數(shù)據(jù)尾:通過(guò)特殊的字符標(biāo)識(shí)#)

  * 相比定長(zhǎng)協(xié)議,如果能夠定義一個(gè)特殊字符作為每個(gè)協(xié)議單元結(jié)束的標(biāo)示,就能夠以變長(zhǎng)的方式進(jìn)行通信,從而在數(shù)據(jù)傳輸和高效之間取得平衡,比如用特殊字符`\n`。特殊結(jié)束符方式的問(wèn)題是過(guò)于簡(jiǎn)單的思考了協(xié)議傳輸?shù)倪^(guò)程,對(duì)于一個(gè)協(xié)議單元必須要全部讀入才能夠進(jìn)行處理,除此之外必須要防止用戶傳輸?shù)臄?shù)據(jù)不能同結(jié)束符相同,否則就會(huì)出現(xiàn)紊亂。

  * 變長(zhǎng)協(xié)議(協(xié)議頭+payload模式)

  * 這種一般是自定義協(xié)議,會(huì)以定長(zhǎng)加不定長(zhǎng)的部分組成,其中定長(zhǎng)的部分需要描述不定長(zhǎng)的內(nèi)容長(zhǎng)度。

  * dubbo就是使用這種形式的數(shù)據(jù)傳輸格式

  Dubbo 框架定義了私有的RPC協(xié)議,其中請(qǐng)求和響應(yīng)協(xié)議的具體內(nèi)容我們使用表格來(lái)展示。

1663309170930_3.jpg

  Dubbo 數(shù)據(jù)包分為消息頭和消息體,消息頭用于存儲(chǔ)一些元信息,比如魔數(shù)(Magic),數(shù)據(jù)包類型(Request/Response),消息體長(zhǎng)度(Data Length)等。消息體中用于存儲(chǔ)具體的調(diào)用消息,比如方法名稱,參數(shù)列表等。下面簡(jiǎn)單列舉一下消息頭的內(nèi)容。

  | 偏移量(Bit) | 字段 | 取值 |

  | ----------- | ------------ | ------------------------------------------------------------ |

  | 0 ~ 7 | 魔數(shù)高位 | 0xda00 |

  | 8 ~ 15 | 魔數(shù)低位 | 0xbb |

  | 16 | 數(shù)據(jù)包類型 | 0 - Response, 1 - Request |

  | 17 | 調(diào)用方式 | 僅在第16位被設(shè)為1的情況下有效,0 - 單向調(diào)用,1 - 雙向調(diào)用 |

  | 18 | 事件標(biāo)識(shí) | 0 - 當(dāng)前數(shù)據(jù)包是請(qǐng)求或響應(yīng)包,1 - 當(dāng)前數(shù)據(jù)包是心跳包 |

  | 19 ~ 23 | 序列化器編號(hào) | 2 - Hessian2Serialization

  3 - JavaSerialization

  4 - CompactedJavaSerialization

  6 - FastJsonSerialization

  7 - NativeJavaSerialization

  8 - KryoSerialization

  9 - FstSerialization |

  | 24 ~ 31 | 狀態(tài) | 20 - OK 30 - CLIENT_TIMEOUT 31 - SERVER_TIMEOUT 40 - BAD_REQUEST 50 - BAD_RESPONSE ...... |

  | 32 ~ 95 | 請(qǐng)求編號(hào) | 共8字節(jié),運(yùn)行時(shí)生成 |

  | 96 ~ 127 | 消息體長(zhǎng)度 | 運(yùn)行時(shí)計(jì)算

  消費(fèi)方發(fā)送請(qǐng)求

  (1)發(fā)送請(qǐng)求

  為了便于大家閱讀代碼,這里以 DemoService 為例,將 sayHello 方法的整個(gè)調(diào)用路徑貼出來(lái)。

proxy0#sayHello(String)
  —> InvokerInvocationHandler#invoke(Object, Method, Object[])
    —> MockClusterInvoker#invoke(Invocation)
      —> AbstractClusterInvoker#invoke(Invocation)
        —> FailoverClusterInvoker#doInvoke(Invocation, List<Invoker<T>>, LoadBalance)
          —> Filter#invoke(Invoker, Invocation)  // 包含多個(gè) Filter 調(diào)用
            —> ListenerInvokerWrapper#invoke(Invocation)
              —> AbstractInvoker#invoke(Invocation)
                —> DubboInvoker#doInvoke(Invocation)
                  —> ReferenceCountExchangeClient#request(Object, int)
                    —> HeaderExchangeClient#request(Object, int)
                      —> HeaderExchangeChannel#request(Object, int)
                        —> AbstractPeer#send(Object)
                          —> AbstractClient#send(Object, boolean)
                            —> NettyChannel#send(Object, boolean)
                              —> NioClientSocketChannel#write(Object)

  dubbo消費(fèi)方,自動(dòng)生成代碼對(duì)象如下

public class proxy0 implements ClassGenerator.DC, EchoService, DemoService {

    private InvocationHandler handler;

    public String sayHello(String string) {
        // 將參數(shù)存儲(chǔ)到 Object 數(shù)組中
        Object[] arrobject = new Object[]{string};
        // 調(diào)用 InvocationHandler 實(shí)現(xiàn)類的 invoke 方法得到調(diào)用結(jié)果
        Object object = this.handler.invoke(this, methods[0], arrobject);
        // 返回調(diào)用結(jié)果
        return (String)object;
    }
}

  InvokerInvocationHandler 中的 invoker 成員變量類型為 MockClusterInvoker,MockClusterInvoker 內(nèi)部封裝了服務(wù)降級(jí)邏輯。下面簡(jiǎn)單看一下:

public Result invoke(Invocation invocation) throws RpcException {
        Result result = null;
        // 獲取 mock 配置值
        String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), MOCK_KEY, Boolean.FALSE.toString()).trim();
        if (value.length() == 0 || value.equalsIgnoreCase("false")) {
             // 無(wú) mock 邏輯,直接調(diào)用其他 Invoker 對(duì)象的 invoke 方法,
            // 比如 FailoverClusterInvoker
            result = this.invoker.invoke(invocation);
        } else if (value.startsWith("force")) {
            // force:xxx 直接執(zhí)行 mock 邏輯,不發(fā)起遠(yuǎn)程調(diào)用
            result = doMockInvoke(invocation, null);
        } else {
             // fail:xxx 表示消費(fèi)方對(duì)調(diào)用服務(wù)失敗后,再執(zhí)行 mock 邏輯,不拋出異常
            try {
                result = this.invoker.invoke(invocation);
            } catch (RpcException e) {
                 // 調(diào)用失敗,執(zhí)行 mock 邏輯
                result = doMockInvoke(invocation, e);
            }
        }
        return result;
    }

  考慮到前文已經(jīng)詳細(xì)分析過(guò) FailoverClusterInvoker,因此本節(jié)略過(guò) FailoverClusterInvoker,直接分析 DubboInvoker。

public abstract class AbstractInvoker<T> implements Invoker<T> {
   
    public Result invoke(Invocation inv) throws RpcException {
        if (destroyed.get()) {
            throw new RpcException("Rpc invoker for service ...");
        }
        RpcInvocation invocation = (RpcInvocation) inv;
        // 設(shè)置 Invoker
        invocation.setInvoker(this);
        if (attachment != null && attachment.size() > 0) {
            // 設(shè)置 attachment
            invocation.addAttachmentsIfAbsent(attachment);
        }
        Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
        if (contextAttachments != null && contextAttachments.size() != 0) {
            // 添加 contextAttachments 到 RpcInvocation#attachment 變量中
            invocation.addAttachments(contextAttachments);
        }
        if (getUrl().getMethodParameter(invocation.getMethodName(), Constants.ASYNC_KEY, false)) {
            // 設(shè)置異步信息到 RpcInvocation#attachment 中
            invocation.setAttachment(Constants.ASYNC_KEY, Boolean.TRUE.toString());
        }
        RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);

        try {
            // 抽象方法,由子類實(shí)現(xiàn)
            return doInvoke(invocation);
        } catch (InvocationTargetException e) {
            // ...
        } catch (RpcException e) {
            // ...
        } catch (Throwable e) {
            return new RpcResult(e);
        }
    }

    protected abstract Result doInvoke(Invocation invocation) throws Throwable;
   
    // 省略其他方法
}

  上面的代碼來(lái)自 AbstractInvoker 類,其中大部分代碼用于添加信息到 RpcInvocation#attachment 變量中,添加完畢后,調(diào)用 doInvoke 執(zhí)行后續(xù)的調(diào)用。doInvoke 是一個(gè)抽象方法,需要由子類實(shí)現(xiàn),下面到 DubboInvoker 中看一下。

@Override
    protected Result doInvoke(final Invocation invocation) throws Throwable {
        RpcInvocation inv = (RpcInvocation) invocation;
        final String methodName = RpcUtils.getMethodName(invocation);
        //將目標(biāo)方法以及版本好作為參數(shù)放入到Invocation中
        inv.setAttachment(PATH_KEY, getUrl().getPath());
        inv.setAttachment(VERSION_KEY, version);

        //獲得客戶端連接
        ExchangeClient currentClient; //初始化invoker的時(shí)候,構(gòu)建的一個(gè)遠(yuǎn)程通信連接
        if (clients.length == 1) { //默認(rèn)
            currentClient = clients[0];
        } else {
            //通過(guò)取模獲得其中一個(gè)連接
            currentClient = clients[index.getAndIncrement() % clients.length];
        }
        try {
            //表示當(dāng)前的方法是否存在返回值
            boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
            int timeout = getUrl().getMethodParameter(methodName, TIMEOUT_KEY, DEFAULT_TIMEOUT);
            //isOneway 為 true,表示“單向”通信
            if (isOneway) {//異步無(wú)返回值
                boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
                currentClient.send(inv, isSent);
                RpcContext.getContext().setFuture(null);
                return AsyncRpcResult.newDefaultAsyncResult(invocation);
            } else { //存在返回值
                //是否采用異步
                AsyncRpcResult asyncRpcResult = new AsyncRpcResult(inv);
                CompletableFuture<Object> responseFuture = currentClient.request(inv, timeout);
                responseFuture.whenComplete((obj, t) -> {
                    if (t != null) {
                        asyncRpcResult.completeExceptionally(t);
                    } else {
                        asyncRpcResult.complete((AppResponse) obj);
                    }
                });
                RpcContext.getContext().setFuture(new FutureAdapter(asyncRpcResult));
                return asyncRpcResult;
            }
        }
        //省略無(wú)關(guān)代碼
    }

  最終進(jìn)入到HeaderExchangeChannel#request方法,拼裝Request并將請(qǐng)求發(fā)送出去

public CompletableFuture<Object> request(Object request, int timeout) throws RemotingException {
        if (closed) {
            throw new RemotingException(this.getLocalAddress(), null, "Failed tosend request " + request + ", cause: The channel " + this + " is closed!");
        }
        // 創(chuàng)建請(qǐng)求對(duì)象
        Request req = new Request();
        req.setVersion(Version.getProtocolVersion());
        req.setTwoWay(true);
        req.setData(request);
        DefaultFuture future = DefaultFuture.newFuture(channel, req, timeout);
        try {
            //NettyClient
            channel.send(req);
        } catch (RemotingException e) {
            future.cancel();
            throw e;
        }
        return future;
    }

  (2)請(qǐng)求編碼

  在netty啟動(dòng)時(shí),我們?cè)O(shè)置了編解碼器,其中通過(guò)ExchangeCodec完成編解碼工作如下:

public class ExchangeCodec extends TelnetCodec {

    // 消息頭長(zhǎng)度
    protected static final int HEADER_LENGTH = 16;
    // 魔數(shù)內(nèi)容
    protected static final short MAGIC = (short) 0xdabb;
    protected static final byte MAGIC_HIGH = Bytes.short2bytes(MAGIC)[0];
    protected static final byte MAGIC_LOW = Bytes.short2bytes(MAGIC)[1];
    protected static final byte FLAG_REQUEST = (byte) 0x80;
    protected static final byte FLAG_TWOWAY = (byte) 0x40;
    protected static final byte FLAG_EVENT = (byte) 0x20;
    protected static final int SERIALIZATION_MASK = 0x1f;
    private static final Logger logger = LoggerFactory.getLogger(ExchangeCodec.class);

    public Short getMagicCode() {
        return MAGIC;
    }

    @Override
    public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws IOException {
        if (msg instanceof Request) {
            // 對(duì) Request 對(duì)象進(jìn)行編碼
            encodeRequest(channel, buffer, (Request) msg);
        } else if (msg instanceof Response) {
            // 對(duì) Response 對(duì)象進(jìn)行編碼,后面分析
            encodeResponse(channel, buffer, (Response) msg);
        } else {
            super.encode(channel, buffer, msg);
        }
    }

    protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException {
        Serialization serialization = getSerialization(channel);

        // 創(chuàng)建消息頭字節(jié)數(shù)組,長(zhǎng)度為 16
        byte[] header = new byte[HEADER_LENGTH];

        // 設(shè)置魔數(shù)
        Bytes.short2bytes(MAGIC, header);

        // 設(shè)置數(shù)據(jù)包類型(Request/Response)和序列化器編號(hào)
        header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId());

        // 設(shè)置通信方式(單向/雙向)
        if (req.isTwoWay()) {
            header[2] |= FLAG_TWOWAY;
        }
       
        // 設(shè)置事件標(biāo)識(shí)
        if (req.isEvent()) {
            header[2] |= FLAG_EVENT;
        }

        // 設(shè)置請(qǐng)求編號(hào),8個(gè)字節(jié),從第4個(gè)字節(jié)開(kāi)始設(shè)置
        Bytes.long2bytes(req.getId(), header, 4);

        // 獲取 buffer 當(dāng)前的寫(xiě)位置
        int savedWriteIndex = buffer.writerIndex();
        // 更新 writerIndex,為消息頭預(yù)留 16 個(gè)字節(jié)的空間
        buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
        ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
        // 創(chuàng)建序列化器,比如 Hessian2ObjectOutput
        ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
        if (req.isEvent()) {
            // 對(duì)事件數(shù)據(jù)進(jìn)行序列化操作
            encodeEventData(channel, out, req.getData());
        } else {
            // 對(duì)請(qǐng)求數(shù)據(jù)進(jìn)行序列化操作
            encodeRequestData(channel, out, req.getData(), req.getVersion());
        }
        out.flushBuffer();
        if (out instanceof Cleanable) {
            ((Cleanable) out).cleanup();
        }
        bos.flush();
        bos.close();
       
        // 獲取寫(xiě)入的字節(jié)數(shù),也就是消息體長(zhǎng)度
        int len = bos.writtenBytes();
        checkPayload(channel, len);

        // 將消息體長(zhǎng)度寫(xiě)入到消息頭中
        Bytes.int2bytes(len, header, 12);

        // 將 buffer 指針移動(dòng)到 savedWriteIndex,為寫(xiě)消息頭做準(zhǔn)備
        buffer.writerIndex(savedWriteIndex);
        // 從 savedWriteIndex 下標(biāo)處寫(xiě)入消息頭
        buffer.writeBytes(header);
        // 設(shè)置新的 writerIndex,writerIndex = 原寫(xiě)下標(biāo) + 消息頭長(zhǎng)度 + 消息體長(zhǎng)度
        buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
    }
   
    // 省略其他方法
}

  以上就是請(qǐng)求對(duì)象的編碼過(guò)程,該過(guò)程首先會(huì)通過(guò)位運(yùn)算將消息頭寫(xiě)入到 header 數(shù)組中。然后對(duì) Request 對(duì)象的 data 字段執(zhí)行序列化操作,序列化后的數(shù)據(jù)最終會(huì)存儲(chǔ)到 ChannelBuffer 中。序列化操作執(zhí)行完后,可得到數(shù)據(jù)序列化后的長(zhǎng)度 len,緊接著將 len 寫(xiě)入到 header 指定位置處。最后再將消息頭字節(jié)數(shù)組 header 寫(xiě)入到 ChannelBuffer 中,整個(gè)編碼過(guò)程就結(jié)束了。本節(jié)的最后,我們?cè)賮?lái)看一下 Request 對(duì)象的 data 字段序列化過(guò)程,也就是 encodeRequestData 方法的邏輯,如下:

public class DubboCodec extends ExchangeCodec implements Codec2 {
   
    protected void encodeRequestData(Channel channel, ObjectOutput out, Object data, String version) throws IOException {
        RpcInvocation inv = (RpcInvocation) data;

        // 依次序列化 dubbo version、path、version
        out.writeUTF(version);
        out.writeUTF(inv.getAttachment(Constants.PATH_KEY));
        out.writeUTF(inv.getAttachment(Constants.VERSION_KEY));

        // 序列化調(diào)用方法名
        out.writeUTF(inv.getMethodName());
        // 將參數(shù)類型轉(zhuǎn)換為字符串,并進(jìn)行序列化
        out.writeUTF(ReflectUtils.getDesc(inv.getParameterTypes()));
        Object[] args = inv.getArguments();
        if (args != null)
            for (int i = 0; i < args.length; i++) {
                // 對(duì)運(yùn)行時(shí)參數(shù)進(jìn)行序列化
                out.writeObject(encodeInvocationArgument(channel, inv, i));
            }
       
        // 序列化 attachments
        out.writeObject(inv.getAttachments());
    }
}

  至此,關(guān)于服務(wù)消費(fèi)方發(fā)送請(qǐng)求的過(guò)程就分析完了,接下來(lái)我們來(lái)看一下服務(wù)提供方是如何接收請(qǐng)求的。

  提供方接收請(qǐng)求

  (1) 請(qǐng)求解碼

  這里直接分析請(qǐng)求數(shù)據(jù)的解碼邏輯,忽略中間過(guò)程,如下:

public class ExchangeCodec extends TelnetCodec {
   
    @Override
    public Object decode(Channel channel, ChannelBuffer buffer) throws IOException {
        int readable = buffer.readableBytes();
        // 創(chuàng)建消息頭字節(jié)數(shù)組
        byte[] header = new byte[Math.min(readable, HEADER_LENGTH)];
        // 讀取消息頭數(shù)據(jù)
        buffer.readBytes(header);
        // 調(diào)用重載方法進(jìn)行后續(xù)解碼工作
        return decode(channel, buffer, readable, header);
    }

    @Override
    protected Object decode(Channel channel, ChannelBuffer buffer, int readable, byte[] header) throws IOException {
        // 檢查魔數(shù)是否相等
        if (readable > 0 && header[0] != MAGIC_HIGH
                || readable > 1 && header[1] != MAGIC_LOW) {
            int length = header.length;
            if (header.length < readable) {
                header = Bytes.copyOf(header, readable);
                buffer.readBytes(header, length, readable - length);
            }
            for (int i = 1; i < header.length - 1; i++) {
                if (header[i] == MAGIC_HIGH && header[i + 1] == MAGIC_LOW) {
                    buffer.readerIndex(buffer.readerIndex() - header.length + i);
                    header = Bytes.copyOf(header, i);
                    break;
                }
            }
            // 通過(guò) telnet 命令行發(fā)送的數(shù)據(jù)包不包含消息頭,所以這里
            // 調(diào)用 TelnetCodec 的 decode 方法對(duì)數(shù)據(jù)包進(jìn)行解碼
            return super.decode(channel, buffer, readable, header);
        }
       
        // 檢測(cè)可讀數(shù)據(jù)量是否少于消息頭長(zhǎng)度,若小于則立即返回 DecodeResult.NEED_MORE_INPUT
        if (readable < HEADER_LENGTH) {
            return DecodeResult.NEED_MORE_INPUT;
        }

        // 從消息頭中獲取消息體長(zhǎng)度
        int len = Bytes.bytes2int(header, 12);
        // 檢測(cè)消息體長(zhǎng)度是否超出限制,超出則拋出異常
        checkPayload(channel, len);

        int tt = len + HEADER_LENGTH;
        // 檢測(cè)可讀的字節(jié)數(shù)是否小于實(shí)際的字節(jié)數(shù)
        if (readable < tt) {
            return DecodeResult.NEED_MORE_INPUT;
        }
       
        ChannelBufferInputStream is = new ChannelBufferInputStream(buffer, len);

        try {
            // 繼續(xù)進(jìn)行解碼工作
            return decodeBody(channel, is, header);
        } finally {
            if (is.available() > 0) {
                try {
                    StreamUtils.skipUnusedStream(is);
                } catch (IOException e) {
                    logger.warn(e.getMessage(), e);
                }
            }
        }
    }
}

  上面方法通過(guò)檢測(cè)消息頭中的魔數(shù)是否與規(guī)定的魔數(shù)相等,提前攔截掉非常規(guī)數(shù)據(jù)包,比如通過(guò) telnet 命令行發(fā)出的數(shù)據(jù)包。接著再對(duì)消息體長(zhǎng)度,以及可讀字節(jié)數(shù)進(jìn)行檢測(cè)。最后調(diào)用 decodeBody 方法進(jìn)行后續(xù)的解碼工作,ExchangeCodec 中實(shí)現(xiàn)了 decodeBody 方法,但因其子類 DubboCodec 覆寫(xiě)了該方法,所以在運(yùn)行時(shí) DubboCodec 中的 decodeBody 方法會(huì)被調(diào)用。下面我們來(lái)看一下該方法的代碼。

public class DubboCodec extends ExchangeCodec implements Codec2 {

    @Override
    protected Object decodeBody(Channel channel, InputStream is, byte[] header) throws IOException {
        // 獲取消息頭中的第三個(gè)字節(jié),并通過(guò)邏輯與運(yùn)算得到序列化器編號(hào)
        byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK);
        Serialization s = CodecSupport.getSerialization(channel.getUrl(), proto);
        // 獲取調(diào)用編號(hào)
        long id = Bytes.bytes2long(header, 4);
        // 通過(guò)邏輯與運(yùn)算得到調(diào)用類型,0 - Response,1 - Request
        if ((flag & FLAG_REQUEST) == 0) {
            // 對(duì)響應(yīng)結(jié)果進(jìn)行解碼,得到 Response 對(duì)象。這個(gè)非本節(jié)內(nèi)容,后面再分析
            // ...
        } else {
            // 創(chuàng)建 Request 對(duì)象
            Request req = new Request(id);
            req.setVersion(Version.getProtocolVersion());
            // 通過(guò)邏輯與運(yùn)算得到通信方式,并設(shè)置到 Request 對(duì)象中
            req.setTwoWay((flag & FLAG_TWOWAY) != 0);
           
            // 通過(guò)位運(yùn)算檢測(cè)數(shù)據(jù)包是否為事件類型
            if ((flag & FLAG_EVENT) != 0) {
                // 設(shè)置心跳事件到 Request 對(duì)象中
                req.setEvent(Request.HEARTBEAT_EVENT);
            }
            try {
                Object data;
                if (req.isHeartbeat()) {
                    // 對(duì)心跳包進(jìn)行解碼,該方法已被標(biāo)注為廢棄
                    data = decodeHeartbeatData(channel, deserialize(s, channel.getUrl(), is));
                } else if (req.isEvent()) {
                    // 對(duì)事件數(shù)據(jù)進(jìn)行解碼
                    data = decodeEventData(channel, deserialize(s, channel.getUrl(), is));
                } else {
                    DecodeableRpcInvocation inv;
                    // 根據(jù) url 參數(shù)判斷是否在 IO 線程上對(duì)消息體進(jìn)行解碼
                    if (channel.getUrl().getParameter(
                            Constants.DECODE_IN_IO_THREAD_KEY,
                            Constants.DEFAULT_DECODE_IN_IO_THREAD)) {
                        inv = new DecodeableRpcInvocation(channel, req, is, proto);
                        // 在當(dāng)前線程,也就是 IO 線程上進(jìn)行后續(xù)的解碼工作。此工作完成后,可將
                        // 調(diào)用方法名、attachment、以及調(diào)用參數(shù)解析出來(lái)
                        inv.decode();
                    } else {
                        // 僅創(chuàng)建 DecodeableRpcInvocation 對(duì)象,但不在當(dāng)前線程上執(zhí)行解碼邏輯
                        inv = new DecodeableRpcInvocation(channel, req,
                                new UnsafeByteArrayInputStream(readMessageData(is)), proto);
                    }
                    data = inv;
                }
               
                // 設(shè)置 data 到 Request 對(duì)象中
                req.setData(data);
            } catch (Throwable t) {
                // 若解碼過(guò)程中出現(xiàn)異常,則將 broken 字段設(shè)為 true,
                // 并將異常對(duì)象設(shè)置到 Reqeust 對(duì)象中
                req.setBroken(true);
                req.setData(t);
            }
            return req;
        }
    }
}

  如上,decodeBody 對(duì)部分字段進(jìn)行了解碼,并將解碼得到的字段封裝到 Request 中。隨后會(huì)調(diào)用 DecodeableRpcInvocation 的 decode 方法進(jìn)行后續(xù)的解碼工作。此工作完成后,可將調(diào)用方法名、attachment、以及調(diào)用參數(shù)解析出來(lái)。

  (2)調(diào)用服務(wù)

  解碼器將數(shù)據(jù)包解析成 Request 對(duì)象后,NettyHandler 的 messageReceived 方法緊接著會(huì)收到這個(gè)對(duì)象,并將這個(gè)對(duì)象繼續(xù)向下傳遞。整個(gè)調(diào)用棧如下:

NettyServerHandler#channelRead(ChannelHandlerContext, MessageEvent)
  —> AbstractPeer#received(Channel, Object)
    —> MultiMessageHandler#received(Channel, Object)
      —> HeartbeatHandler#received(Channel, Object)
        —> AllChannelHandler#received(Channel, Object)
          —> ExecutorService#execute(Runnable)    // 由線程池執(zhí)行后續(xù)的調(diào)用邏輯

  這里我們直接分析調(diào)用棧中的分析第一個(gè)和最后一個(gè)調(diào)用方法邏輯。如下:

  考慮到篇幅,以及很多中間調(diào)用的邏輯并非十分重要,所以這里就不對(duì)調(diào)用棧中的每個(gè)方法都進(jìn)行分析了。這里我們直接分析最后一個(gè)調(diào)用方法邏輯。如下:

public class ChannelEventRunnable implements Runnable {
   
    private final ChannelHandler handler;
    private final Channel channel;
    private final ChannelState state;
    private final Throwable exception;
    private final Object message;
   
    @Override
    public void run() {
        // 檢測(cè)通道狀態(tài),對(duì)于請(qǐng)求或響應(yīng)消息,此時(shí) state = RECEIVED
        if (state == ChannelState.RECEIVED) {
            try {
                // 將 channel 和 message 傳給 ChannelHandler 對(duì)象,進(jìn)行后續(xù)的調(diào)用
                handler.received(channel, message);
            } catch (Exception e) {
                logger.warn("... operation error, channel is ... message is ...");
            }
        }
       
        // 其他消息類型通過(guò) switch 進(jìn)行處理
        else {
            switch (state) {
            case CONNECTED:
                try {
                    handler.connected(channel);
                } catch (Exception e) {
                    logger.warn("... operation error, channel is ...");
                }
                break;
            case DISCONNECTED:
                // ...
            case SENT:
                // ...
            case CAUGHT:
                // ...
            default:
                logger.warn("unknown state: " + state + ", message is " + message);
            }
        }

    }
}

  如上,請(qǐng)求和響應(yīng)消息出現(xiàn)頻率明顯比其他類型消息高,所以這里對(duì)該類型的消息進(jìn)行了針對(duì)性判斷。ChannelEventRunnable 僅是一個(gè)中轉(zhuǎn)站,它的 run 方法中并不包含具體的調(diào)用邏輯,僅用于將參數(shù)傳給其他 ChannelHandler 對(duì)象進(jìn)行處理,該對(duì)象類型為 DecodeHandler。

public class DecodeHandler extends AbstractChannelHandlerDelegate {

    public DecodeHandler(ChannelHandler handler) {
        super(handler);
    }

    @Override
    public void received(Channel channel, Object message) throws RemotingException {
        if (message instanceof Decodeable) {
            // 對(duì) Decodeable 接口實(shí)現(xiàn)類對(duì)象進(jìn)行解碼
            decode(message);
        }

        if (message instanceof Request) {
            // 對(duì) Request 的 data 字段進(jìn)行解碼
            decode(((Request) message).getData());
        }

        if (message instanceof Response) {
            // 對(duì) Request 的 result 字段進(jìn)行解碼
            decode(((Response) message).getResult());
        }

        // 執(zhí)行后續(xù)邏輯
        handler.received(channel, message);
    }

    private void decode(Object message) {
        // Decodeable 接口目前有兩個(gè)實(shí)現(xiàn)類,
        // 分別為 DecodeableRpcInvocation 和 DecodeableRpcResult
        if (message != null && message instanceof Decodeable) {
            try {
                // 執(zhí)行解碼邏輯
                ((Decodeable) message).decode();
            } catch (Throwable e) {
                if (log.isWarnEnabled()) {
                    log.warn("Call Decodeable.decode failed: " + e.getMessage(), e);
                }
            }
        }
    }
}

  DecodeHandler 主要是包含了一些解碼邏輯,完全解碼后的 Request 對(duì)象會(huì)繼續(xù)向后傳遞

public class DubboProtocol extends AbstractProtocol {

    public static final String NAME = "dubbo";
   
    private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {

        @Override
        public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
            if (message instanceof Invocation) {
                Invocation inv = (Invocation) message;
                // 獲取 Invoker 實(shí)例
                Invoker<?> invoker = getInvoker(channel, inv);
                if (Boolean.TRUE.toString().equals(inv.getAttachments().get(IS_CALLBACK_SERVICE_INVOKE))) {
                    // 回調(diào)相關(guān),忽略
                }
                RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
                // 通過(guò) Invoker 調(diào)用具體的服務(wù)
                return invoker.invoke(inv);
            }
            throw new RemotingException(channel, "Unsupported request: ...");
        }
       
        // 忽略其他方法
    }
   
    Invoker<?> getInvoker(Channel channel, Invocation inv) throws RemotingException {
        // 忽略回調(diào)和本地存根相關(guān)邏輯
        // ...
       
        int port = channel.getLocalAddress().getPort();
       
        // 計(jì)算 service key,格式為 groupName/serviceName:serviceVersion:port。比如:
        //   dubbo/com.alibaba.dubbo.demo.DemoService:1.0.0:20880
        String serviceKey = serviceKey(port, path, inv.getAttachments().get(Constants.VERSION_KEY), inv.getAttachments().get(Constants.GROUP_KEY));

        // 從 exporterMap 查找與 serviceKey 相對(duì)應(yīng)的 DubboExporter 對(duì)象,
        // 服務(wù)導(dǎo)出過(guò)程中會(huì)將 <serviceKey, DubboExporter> 映射關(guān)系存儲(chǔ)到 exporterMap 集合中
        DubboExporter<?> exporter = (DubboExporter<?>) exporterMap.get(serviceKey);

        if (exporter == null)
            throw new RemotingException(channel, "Not found exported service ...");

        // 獲取 Invoker 對(duì)象,并返回
        return exporter.getInvoker();
    }
   
    // 忽略其他方法
}

  在之前課程中介紹過(guò),服務(wù)全部暴露完成之后保存到exporterMap中。這里就是通過(guò)serviceKey獲取exporter之后獲取Invoker,并通過(guò) Invoker 的 invoke 方法調(diào)用服務(wù)邏輯

public abstract class AbstractProxyInvoker<T> implements Invoker<T> {

    @Override
    public Result invoke(Invocation invocation) throws RpcException {
        try {
            // 調(diào)用 doInvoke 執(zhí)行后續(xù)的調(diào)用,并將調(diào)用結(jié)果封裝到 RpcResult 中,并
            return new RpcResult(doInvoke(proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments()));
        } catch (InvocationTargetException e) {
            return new RpcResult(e.getTargetException());
        } catch (Throwable e) {
            throw new RpcException("Failed to invoke remote proxy method ...");
        }
    }
   
    protected abstract Object doInvoke(T proxy, String methodName, Class<?>[] parameterTypes, Object[] arguments) throws Throwable;
}

  如上,doInvoke 是一個(gè)抽象方法,這個(gè)需要由具體的 Invoker 實(shí)例實(shí)現(xiàn)。Invoker 實(shí)例是在運(yùn)行時(shí)通過(guò) JavassistProxyFactory 創(chuàng)建的,創(chuàng)建邏輯如下:

public class JavassistProxyFactory extends AbstractProxyFactory {
   
    // 省略其他方法

    @Override
    public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
        final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
        // 創(chuàng)建匿名類對(duì)象
        return new AbstractProxyInvoker<T>(proxy, type, url) {
            @Override
            protected Object doInvoke(T proxy, String methodName,
                                      Class<?>[] parameterTypes,
                                      Object[] arguments) throws Throwable {
                // 調(diào)用 invokeMethod 方法進(jìn)行后續(xù)的調(diào)用
                return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
            }
        };
    }
}

  Wrapper 是一個(gè)抽象類,其中 invokeMethod 是一個(gè)抽象方法。Dubbo 會(huì)在運(yùn)行時(shí)通過(guò) Javassist 框架為 Wrapper 生成實(shí)現(xiàn)類,并實(shí)現(xiàn) invokeMethod 方法,該方法最終會(huì)根據(jù)調(diào)用信息調(diào)用具體的服務(wù)。以 DemoServiceImpl 為例,Javassist 為其生成的代理類如下。

/** Wrapper0 是在運(yùn)行時(shí)生成的,大家可使用 Arthas 進(jìn)行反編譯 */
public class Wrapper0 extends Wrapper implements ClassGenerator.DC {
    public static String[] pns;
    public static Map pts;
    public static String[] mns;
    public static String[] dmns;
    public static Class[] mts0;

    // 省略其他方法

    public Object invokeMethod(Object object, String string, Class[] arrclass, Object[] arrobject) throws InvocationTargetException {
        DemoService demoService;
        try {
            // 類型轉(zhuǎn)換
            demoService = (DemoService)object;
        }
        catch (Throwable throwable) {
            throw new IllegalArgumentException(throwable);
        }
        try {
            // 根據(jù)方法名調(diào)用指定的方法
            if ("sayHello".equals(string) && arrclass.length == 1) {
                return demoService.sayHello((String)arrobject[0]);
            }
        }
        catch (Throwable throwable) {
            throw new InvocationTargetException(throwable);
        }
        throw new NoSuchMethodException(new StringBuffer().append("Not found method \"").append(string).append("\" in class com.alibaba.dubbo.demo.DemoService.").toString());
    }
}

  到這里,整個(gè)服務(wù)調(diào)用過(guò)程就分析完了。最后把調(diào)用過(guò)程貼出來(lái),如下:

ChannelEventRunnable#run()
  —> DecodeHandler#received(Channel, Object)
    —> HeaderExchangeHandler#received(Channel, Object)
      —> HeaderExchangeHandler#handleRequest(ExchangeChannel, Request)
        —> DubboProtocol.requestHandler#reply(ExchangeChannel, Object)
          —> Filter#invoke(Invoker, Invocation)
            —> AbstractProxyInvoker#invoke(Invocation)
              —> Wrapper0#invokeMethod(Object, String, Class[], Object[])
                —> DemoServiceImpl#sayHello(String)

  提供方返回調(diào)用結(jié)果

  服務(wù)提供方調(diào)用指定服務(wù)后,會(huì)將調(diào)用結(jié)果封裝到 Response 對(duì)象中,并將該對(duì)象返回給服務(wù)消費(fèi)方。服務(wù)提供方也是通過(guò) NettyChannel 的 send 方法將 Response 對(duì)象返回,這里就不在重復(fù)分析了。本節(jié)我們僅需關(guān)注 Response 對(duì)象的編碼過(guò)程即可。

public class ExchangeCodec extends TelnetCodec {
    public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws IOException {
        if (msg instanceof Request) {
            encodeRequest(channel, buffer, (Request) msg);
        } else if (msg instanceof Response) {
            // 對(duì)響應(yīng)對(duì)象進(jìn)行編碼
            encodeResponse(channel, buffer, (Response) msg);
        } else {
            super.encode(channel, buffer, msg);
        }
    }
   
    protected void encodeResponse(Channel channel, ChannelBuffer buffer, Response res) throws IOException {
        int savedWriteIndex = buffer.writerIndex();
        try {
            Serialization serialization = getSerialization(channel);
            // 創(chuàng)建消息頭字節(jié)數(shù)組
            byte[] header = new byte[HEADER_LENGTH];
            // 設(shè)置魔數(shù)
            Bytes.short2bytes(MAGIC, header);
            // 設(shè)置序列化器編號(hào)
            header[2] = serialization.getContentTypeId();
            if (res.isHeartbeat()) header[2] |= FLAG_EVENT;
            // 獲取響應(yīng)狀態(tài)
            byte status = res.getStatus();
            // 設(shè)置響應(yīng)狀態(tài)
            header[3] = status;
            // 設(shè)置請(qǐng)求編號(hào)
            Bytes.long2bytes(res.getId(), header, 4);

            // 更新 writerIndex,為消息頭預(yù)留 16 個(gè)字節(jié)的空間
            buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
            ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
            ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
           
            if (status == Response.OK) {
                if (res.isHeartbeat()) {
                    // 對(duì)心跳響應(yīng)結(jié)果進(jìn)行序列化,已廢棄
                    encodeHeartbeatData(channel, out, res.getResult());
                } else {
                    // 對(duì)調(diào)用結(jié)果進(jìn)行序列化
                    encodeResponseData(channel, out, res.getResult(), res.getVersion());
                }
            } else {
                // 對(duì)錯(cuò)誤信息進(jìn)行序列化
                out.writeUTF(res.getErrorMessage())
            };
            out.flushBuffer();
            if (out instanceof Cleanable) {
                ((Cleanable) out).cleanup();
            }
            bos.flush();
            bos.close();

            // 獲取寫(xiě)入的字節(jié)數(shù),也就是消息體長(zhǎng)度
            int len = bos.writtenBytes();
            checkPayload(channel, len);
           
            // 將消息體長(zhǎng)度寫(xiě)入到消息頭中
            Bytes.int2bytes(len, header, 12);
            // 將 buffer 指針移動(dòng)到 savedWriteIndex,為寫(xiě)消息頭做準(zhǔn)備
            buffer.writerIndex(savedWriteIndex);
            // 從 savedWriteIndex 下標(biāo)處寫(xiě)入消息頭
            buffer.writeBytes(header);
            // 設(shè)置新的 writerIndex,writerIndex = 原寫(xiě)下標(biāo) + 消息頭長(zhǎng)度 + 消息體長(zhǎng)度
            buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
        } catch (Throwable t) {
            // 異常處理邏輯不是很難理解,但是代碼略多,這里忽略了
        }
    }
}

public class DubboCodec extends ExchangeCodec implements Codec2 {
   
    protected void encodeResponseData(Channel channel, ObjectOutput out, Object data, String version) throws IOException {
        Result result = (Result) data;
        // 檢測(cè)當(dāng)前協(xié)議版本是否支持帶有 attachment 集合的 Response 對(duì)象
        boolean attach = Version.isSupportResponseAttachment(version);
        Throwable th = result.getException();
       
        // 異常信息為空
        if (th == null) {
            Object ret = result.getValue();
            // 調(diào)用結(jié)果為空
            if (ret == null) {
                // 序列化響應(yīng)類型
                out.writeByte(attach ? RESPONSE_NULL_VALUE_WITH_ATTACHMENTS : RESPONSE_NULL_VALUE);
            }
            // 調(diào)用結(jié)果非空
            else {
                // 序列化響應(yīng)類型
                out.writeByte(attach ? RESPONSE_VALUE_WITH_ATTACHMENTS : RESPONSE_VALUE);
                // 序列化調(diào)用結(jié)果
                out.writeObject(ret);
            }
        }
        // 異常信息非空
        else {
            // 序列化響應(yīng)類型
            out.writeByte(attach ? RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS : RESPONSE_WITH_EXCEPTION);
            // 序列化異常對(duì)象
            out.writeObject(th);
        }

        if (attach) {
            // 記錄 Dubbo 協(xié)議版本
            result.getAttachments().put(Constants.DUBBO_VERSION_KEY, Version.getProtocolVersion());
            // 序列化 attachments 集合
            out.writeObject(result.getAttachments());
        }
    }
}

  以上就是 Response 對(duì)象編碼的過(guò)程,和前面分析的 Request 對(duì)象編碼過(guò)程很相似。如果大家能看 Request 對(duì)象的編碼邏輯,那么這里的 Response 對(duì)象的編碼邏輯也不難理解,就不多說(shuō)了。接下來(lái)我們?cè)賮?lái)分析雙向通信的最后一環(huán) —— 服務(wù)消費(fèi)方接收調(diào)用結(jié)果。

  消費(fèi)方接收調(diào)用結(jié)果

  服務(wù)消費(fèi)方在收到響應(yīng)數(shù)據(jù)后,首先要做的事情是對(duì)響應(yīng)數(shù)據(jù)進(jìn)行解碼,得到 Response 對(duì)象。然后再將該對(duì)象傳遞給下一個(gè)入站處理器,這個(gè)入站處理器就是 NettyHandler。接下來(lái) NettyHandler 會(huì)將這個(gè)對(duì)象繼續(xù)向下傳遞,最后 AllChannelHandler 的 received 方法會(huì)收到這個(gè)對(duì)象,并將這個(gè)對(duì)象派發(fā)到線程池中。這個(gè)過(guò)程和服務(wù)提供方接收請(qǐng)求的過(guò)程是一樣的,因此這里就不重復(fù)分析了。

  (1)響應(yīng)數(shù)據(jù)解碼

  響應(yīng)數(shù)據(jù)解碼邏輯主要的邏輯封裝在 DubboCodec 中,我們直接分析這個(gè)類的代碼。如下:

public class DubboCodec extends ExchangeCodec implements Codec2 {

    @Override
    protected Object decodeBody(Channel channel, InputStream is, byte[] header) throws IOException {
        byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK);
        Serialization s = CodecSupport.getSerialization(channel.getUrl(), proto);
        // 獲取請(qǐng)求編號(hào)
        long id = Bytes.bytes2long(header, 4);
        // 檢測(cè)消息類型,若下面的條件成立,表明消息類型為 Response
        if ((flag & FLAG_REQUEST) == 0) {
            // 創(chuàng)建 Response 對(duì)象
            Response res = new Response(id);
            // 檢測(cè)事件標(biāo)志位
            if ((flag & FLAG_EVENT) != 0) {
                // 設(shè)置心跳事件
                res.setEvent(Response.HEARTBEAT_EVENT);
            }
            // 獲取響應(yīng)狀態(tài)
            byte status = header[3];
            // 設(shè)置響應(yīng)狀態(tài)
            res.setStatus(status);
           
            // 如果響應(yīng)狀態(tài)為 OK,表明調(diào)用過(guò)程正常
            if (status == Response.OK) {
                try {
                    Object data;
                    if (res.isHeartbeat()) {
                        // 反序列化心跳數(shù)據(jù),已廢棄
                        data = decodeHeartbeatData(channel, deserialize(s, channel.getUrl(), is));
                    } else if (res.isEvent()) {
                        // 反序列化事件數(shù)據(jù)
                        data = decodeEventData(channel, deserialize(s, channel.getUrl(), is));
                    } else {
                        DecodeableRpcResult result;
                        // 根據(jù) url 參數(shù)決定是否在 IO 線程上執(zhí)行解碼邏輯
                        if (channel.getUrl().getParameter(
                                Constants.DECODE_IN_IO_THREAD_KEY,
                                Constants.DEFAULT_DECODE_IN_IO_THREAD)) {
                            // 創(chuàng)建 DecodeableRpcResult 對(duì)象
                            result = new DecodeableRpcResult(channel, res, is,
                                    (Invocation) getRequestData(id), proto);
                            // 進(jìn)行后續(xù)的解碼工作
                            result.decode();
                        } else {
                            // 創(chuàng)建 DecodeableRpcResult 對(duì)象
                            result = new DecodeableRpcResult(channel, res,
                                    new UnsafeByteArrayInputStream(readMessageData(is)),
                                    (Invocation) getRequestData(id), proto);
                        }
                        data = result;
                    }
                   
                    // 設(shè)置 DecodeableRpcResult 對(duì)象到 Response 對(duì)象中
                    res.setResult(data);
                } catch (Throwable t) {
                    // 解碼過(guò)程中出現(xiàn)了錯(cuò)誤,此時(shí)設(shè)置 CLIENT_ERROR 狀態(tài)碼到 Response 對(duì)象中
                    res.setStatus(Response.CLIENT_ERROR);
                    res.setErrorMessage(StringUtils.toString(t));
                }
            }
            // 響應(yīng)狀態(tài)非 OK,表明調(diào)用過(guò)程出現(xiàn)了異常
            else {
                // 反序列化異常信息,并設(shè)置到 Response 對(duì)象中
                res.setErrorMessage(deserialize(s, channel.getUrl(), is).readUTF());
            }
            return res;
        } else {
            // 對(duì)請(qǐng)求數(shù)據(jù)進(jìn)行解碼,前面已分析過(guò),此處忽略
        }
    }
}

  以上就是響應(yīng)數(shù)據(jù)的解碼過(guò)程,上面邏輯看起來(lái)是不是似曾相識(shí)。對(duì)的,我們?cè)谇懊嬲鹿?jié)分析過(guò) DubboCodec 的 decodeBody 方法中關(guān)于請(qǐng)求數(shù)據(jù)的解碼過(guò)程,該過(guò)程和響應(yīng)數(shù)據(jù)的解碼過(guò)程很相似。下面,我們繼續(xù)分析調(diào)用結(jié)果的反序列化過(guò)程

public class DecodeableRpcResult extends AppResponse implements Codec, Decodeable {

    private static final Logger log = LoggerFactory.getLogger(DecodeableRpcResult.class);

    private Channel channel;

    private byte serializationType;

    private InputStream inputStream;

    private Response response;

    private Invocation invocation;

    private volatile boolean hasDecoded;

    public DecodeableRpcResult(Channel channel, Response response, InputStream is, Invocation invocation, byte id) {
        Assert.notNull(channel, "channel == null");
        Assert.notNull(response, "response == null");
        Assert.notNull(is, "inputStream == null");
        this.channel = channel;
        this.response = response;
        this.inputStream = is;
        this.invocation = invocation;
        this.serializationType = id;
    }

    @Override
    public void encode(Channel channel, OutputStream output, Object message) throws IOException {
        throw new UnsupportedOperationException();
    }

    @Override
    public Object decode(Channel channel, InputStream input) throws IOException {
        ObjectInput in = CodecSupport.getSerialization(channel.getUrl(), serializationType)
                .deserialize(channel.getUrl(), input);
        // 反序列化響應(yīng)類型
        byte flag = in.readByte();
        switch (flag) {
            case DubboCodec.RESPONSE_NULL_VALUE:
                break;
            case DubboCodec.RESPONSE_VALUE:
                handleValue(in);
                break;
            case DubboCodec.RESPONSE_WITH_EXCEPTION:
                handleException(in);
                break;
                // 返回值為空,且攜帶了 attachments 集合
            case DubboCodec.RESPONSE_NULL_VALUE_WITH_ATTACHMENTS:
                handleAttachment(in);
                break;
                //返回值不為空,且攜帶了 attachments 集合
            case DubboCodec.RESPONSE_VALUE_WITH_ATTACHMENTS:
                handleValue(in);
                handleAttachment(in);
                break;
            // 異常對(duì)象不為空,且攜帶了 attachments 集合
            case DubboCodec.RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS:
                handleException(in);
                handleAttachment(in);
                break;
            default:
                throw new IOException("Unknown result flag, expect '0' '1' '2' '3' '4' '5', but received: " + flag);
        }
        if (in instanceof Cleanable) {
            ((Cleanable) in).cleanup();
        }
        return this;
    }

  正常調(diào)用下,線程會(huì)進(jìn)入 RESPONSE_VALUE_WITH_ATTACHMENTS 分支中。然后線程會(huì)從 invocation 變量(大家探索一下 invocation 變量的由來(lái))中獲取返回值類型,接著對(duì)調(diào)用結(jié)果進(jìn)行反序列化,并將序列化后的結(jié)果存儲(chǔ)起來(lái)。最后對(duì) attachments 集合進(jìn)行反序列化,并存到指定字段中。

  異步轉(zhuǎn)同步

  Dubbo發(fā)送數(shù)據(jù)至服務(wù)方后,在通信層面是異步的,通信線程并不會(huì)等待結(jié)果數(shù)據(jù)返回。而我們?cè)谑褂肈ubbo進(jìn)行RPC調(diào)用缺省就是同步的,這其中就涉及到了異步轉(zhuǎn)同步的操作。

  而在2.7.x版本中,這種自實(shí)現(xiàn)的異步轉(zhuǎn)同步操作進(jìn)行了修改。新的`DefaultFuture`繼承了`CompletableFuture`,新的`doReceived(Response res)`方法如下:

private void doReceived(Response res) {
    if (res == null) {
        throw new IllegalStateException("response cannot be null");
    }
    if (res.getStatus() == Response.OK) {
        this.complete(res.getResult());
    } else if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) {
        this.completeExceptionally(new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, channel, res.getErrorMessage()));
    } else {
        this.completeExceptionally(new RemotingException(channel, res.getErrorMessage()));
    }
}

  通過(guò)`CompletableFuture#complete`方法來(lái)設(shè)置異步的返回結(jié)果,且刪除舊的`get()`方法,使用`CompletableFuture#get()`方法:

public T get() throws InterruptedException, ExecutionException {
    Object r;
    return reportGet((r = result) == null ? waitingGet(true) : r);
}

  使用`CompletableFuture`完成了異步轉(zhuǎn)同步的操作。

  異步多線程數(shù)據(jù)一致

  這里簡(jiǎn)單說(shuō)明一下。一般情況下,服務(wù)消費(fèi)方會(huì)并發(fā)調(diào)用多個(gè)服務(wù),每個(gè)用戶線程發(fā)送請(qǐng)求后,會(huì)調(diào)用 get 方法進(jìn)行等待。 一段時(shí)間后,服務(wù)消費(fèi)方的線程池會(huì)收到多個(gè)響應(yīng)對(duì)象。這個(gè)時(shí)候要考慮一個(gè)問(wèn)題,如何將每個(gè)響應(yīng)對(duì)象傳遞給相應(yīng)的 Future 對(duì)象,不出錯(cuò)。答案是通過(guò)調(diào)用**編號(hào)**。Future 被創(chuàng)建時(shí),會(huì)要求傳入一個(gè) Request 對(duì)象。此時(shí) DefaultFuture 可從 Request 對(duì)象中獲取調(diào)用編號(hào),并將 <調(diào)用編號(hào), DefaultFuture 對(duì)象> 映射關(guān)系存入到靜態(tài) Map 中,即 FUTURES。線程池中的線程在收到 Response 對(duì)象后,會(huì)根據(jù) Response 對(duì)象中的調(diào)用編號(hào)到 FUTURES 集合中取出相應(yīng)的 DefaultFuture 對(duì)象,然后再將 Response 對(duì)象設(shè)置到 DefaultFuture 對(duì)象中。這樣用戶線程即可從 DefaultFuture 對(duì)象中獲取調(diào)用結(jié)果了。整個(gè)過(guò)程大致如下圖:

  

1663310007996_4.jpg

private DefaultFuture(Channel channel, Request request, int timeout) {
    this.channel = channel;
    this.request = request;
    this.id = request.getId();
    this.timeout = timeout > 0 ? timeout : channel.getUrl().getPositiveParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT);
    // put into waiting map.
    FUTURES.put(id, this);
    CHANNELS.put(id, channel);
}

  心跳檢查

  Dubbo采用雙向心跳的方式檢測(cè)Client端與Server端的連通性。

  我們?cè)賮?lái)看看 Dubbo 是如何設(shè)計(jì)應(yīng)用層心跳的。Dubbo 的心跳是雙向心跳,客戶端會(huì)給服務(wù)端發(fā)送心跳,反之,服務(wù)端也會(huì)向客戶端發(fā)送心跳。

  創(chuàng)建定時(shí)器

public class HeaderExchangeClient implements ExchangeClient {

    private final Client client;
    private final ExchangeChannel channel;

    private static final HashedWheelTimer IDLE_CHECK_TIMER = new HashedWheelTimer(
            new NamedThreadFactory("dubbo-client-idleCheck", true), 1, TimeUnit.SECONDS, TICKS_PER_WHEEL);
   
    private HeartbeatTimerTask heartBeatTimerTask;
    private ReconnectTimerTask reconnectTimerTask;

    public HeaderExchangeClient(Client client, boolean startTimer) {
        Assert.notNull(client, "Client can't be null");
        this.client = client;
        this.channel = new HeaderExchangeChannel(client);

        if (startTimer) {
            URL url = client.getUrl();
            //開(kāi)啟心跳失敗之后處理重連,斷連的邏輯定時(shí)任務(wù)
            startReconnectTask(url);
            //開(kāi)啟發(fā)送心跳請(qǐng)求定時(shí)任務(wù)
            startHeartBeatTask(url);
        }
    }

  Dubbo 在 `HeaderExchangeClient `初始化時(shí)開(kāi)啟了兩個(gè)定時(shí)任務(wù)

  `startReconnectTask` 主要用于定時(shí)發(fā)送心跳請(qǐng)求

  `startHeartBeatTask` 主要用于心跳失敗之后處理重連,斷連的邏輯

  發(fā)送心跳請(qǐng)求

  詳細(xì)解析下心跳檢測(cè)定時(shí)任務(wù)的邏輯 `HeartbeatTimerTask#doTask`:

protected void doTask(Channel channel) {
      Long lastRead = lastRead(channel);
      Long lastWrite = lastWrite(channel);
      if ((lastRead != null && now() - lastRead > heartbeat)
          || (lastWrite != null && now() - lastWrite > heartbeat)) {
          Request req = new Request();
          req.setVersion(Version.getProtocolVersion());
          req.setTwoWay(true);
          req.setEvent(Request.HEARTBEAT_EVENT);
          channel.send(req);
      }
   }

  前面已經(jīng)介紹過(guò),**Dubbo 采取的是雙向心跳設(shè)計(jì)**,即服務(wù)端會(huì)向客戶端發(fā)送心跳,客戶端也會(huì)向服務(wù)端發(fā)送心跳,接收的一方更新 lastRead 字段,發(fā)送的一方更新 lastWrite 字段,超過(guò)心跳間隙的時(shí)間,便發(fā)送心跳請(qǐng)求給對(duì)端。這里的 lastRead/lastWrite 同樣會(huì)被同一個(gè)通道上的普通調(diào)用更新,通過(guò)更新這兩個(gè)字段,實(shí)現(xiàn)了只在連接空閑時(shí)才會(huì)真正發(fā)送空閑報(bào)文的機(jī)制,符合我們一開(kāi)始科普的做法。

  處理重連和斷連

  繼續(xù)研究下重連和斷連定時(shí)器都實(shí)現(xiàn)了什么 `ReconnectTimerTask#doTask`。

   protected void doTask(Channel channel) {
       Long lastRead = lastRead(channel);
       Long now = now();
       if (!channel.isConnected()) {
           ((Client) channel).reconnect();
           // check pong at client
       } else if (lastRead != null && now - lastRead > idleTimeout) {
           ((Client) channel).reconnect();
       }
    }

  第二個(gè)定時(shí)器則負(fù)責(zé)根據(jù)客戶端、服務(wù)端類型來(lái)對(duì)連接做不同的處理,當(dāng)超過(guò)設(shè)置的心跳總時(shí)間之后,客戶端選擇的是重新連接,服務(wù)端則是選擇直接斷開(kāi)連接。這樣的考慮是合理的,客戶端調(diào)用是強(qiáng)依賴可用連接的,而服務(wù)端可以等待客戶端重新建立連接。

  Dubbo 對(duì)于建立的每一個(gè)連接,同時(shí)在客戶端和服務(wù)端開(kāi)啟了 2 個(gè)定時(shí)器,一個(gè)用于定時(shí)發(fā)送心跳,一個(gè)用于定時(shí)重連、斷連,執(zhí)行的頻率均為各自檢測(cè)周期的 1/3。定時(shí)發(fā)送心跳的任務(wù)負(fù)責(zé)在連接空閑時(shí),向?qū)Χ税l(fā)送心跳包。定時(shí)重連、斷連的任務(wù)負(fù)責(zé)檢測(cè) lastRead 是否在超時(shí)周期內(nèi)仍未被更新,如果判定為超時(shí),客戶端處理的邏輯是重連,服務(wù)端則采取斷連的措施。

分享到:
在線咨詢 我要報(bào)名
和我們?cè)诰€交談!