更新時(shí)間:2022-09-16 來源:黑馬程序員 瀏覽量:
在之前的內(nèi)容中，我們講解了消費(fèi)者端服務(wù)發(fā)現(xiàn)與提供者端服務(wù)暴露的相關(guān)內(nèi)容，同時(shí)也知道消費(fèi)者端通過內(nèi)置的負(fù)載均衡算法獲取合適的調(diào)用invoker進(jìn)行遠(yuǎn)程調(diào)用。那么，本章節(jié)重點(diǎn)關(guān)注的就是遠(yuǎn)程調(diào)用過程即網(wǎng)絡(luò)通信。
網(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 語義;
網(wǎng)絡(luò)通信的問題：
客戶端與服務(wù)端連通性問題
粘包拆包問題
異步多線程數(shù)據(jù)一致問題
通信協(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)者無法壓滿提供者，盡量不要用 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)景：頁面?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)集成，跨語言調(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)寫)實(shí)現(xiàn)的REST調(diào)用支持
grpc協(xié)議
Dubbo 自 2.7.5 版本開始支持 gRPC 協(xié)議，對(duì)于計(jì)劃使用 HTTP/2 通信，或者想利用 gRPC 帶來的 Stream、反壓、Reactive 編程等能力的開發(fā)者來說， 都可以考慮啟用 gRPC 協(xié)議。
- 為期望使用 gRPC 協(xié)議的用戶帶來服務(wù)治理能力，方便接入 Dubbo 體系
- 用戶可以使用 Dubbo 風(fēng)格的，基于接口的編程風(fēng)格來定義和使用遠(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)類
JavaEE
鴻蒙應(yīng)用開發(fā)
集成電路應(yīng)用開發(fā)
人工智能開發(fā)
AI+Linux云計(jì)算運(yùn)維
Python+大數(shù)據(jù)開發(fā)
AI+設(shè)計(jì)
軟件測(cè)試
新媒體+短視頻
HTML&JS+前端
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序列化接口。
Dubbo序列化支持java、compactedjava、nativejava、fastjson、dubbo、fst、hessian2、kryo，protostuff其中默認(rèn)hessian2。其中java、compactedjava、nativejava屬于原生java的序列化。
- dubbo序列化：阿里尚未開發(fā)成熟的高效java序列化實(shí)現(xiàn)，阿里不建議在生產(chǎn)環(huán)境使用它。
- **hessian2序列化：hessian是一種跨語言的高效二進(jìn)制序列化方式。但這里實(shí)際不是原生的hessian2序列化，而是阿里修改過的，它是dubbo RPC默認(rèn)啟用的序列化方式。**
- json序列化：目前有兩種實(shí)現(xiàn)，一種是采用的阿里的fastjson庫，另一種是采用dubbo中自己實(shí)現(xiàn)的簡(jiǎn)單json庫，但其實(shí)現(xiàn)都不是特別成熟，而且json這種文本序列化性能一般不如上面兩種二進(jìn)制序列化。
- java序列化：主要是采用JDK自帶的Java序列化實(shí)現(xiàn)，性能很不理想。
最近幾年，各種新的高效序列化方式層出不窮，不斷刷新序列化性能的上限，最典型的包括：
- 專門針對(duì)Java語言的：Kryo，F(xiàn)ST等等
- 跨語言的：Protostuff，ProtoBuf，Thrift，Avro，MsgPack等等
這些序列化方式的性能多數(shù)都顯著優(yōu)于 hessian2 (甚至包括尚未成熟的dubbo序列化)。所以我們可以為 dubbo 引入 Kryo 和 FST 這兩種高效 Java 來優(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ù)粘包拆包問題，一般有三種方式
* 定長(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é)議在讀取或者寫入時(shí)，效率比較高，因?yàn)閿?shù)據(jù)緩存的大小基本都確定了，就好比數(shù)組一樣，缺陷就是適應(yīng)性不足，以RPC場(chǎng)景為例，很難估計(jì)出定長(zhǎng)的長(zhǎng)度是多少。
* 特殊結(jié)束符(數(shù)據(jù)尾：通過特殊的字符標(biāo)識(shí)#)
* 相比定長(zhǎng)協(xié)議，如果能夠定義一個(gè)特殊字符作為每個(gè)協(xié)議單元結(jié)束的標(biāo)示，就能夠以變長(zhǎng)的方式進(jìn)行通信，從而在數(shù)據(jù)傳輸和高效之間取得平衡，比如用特殊字符`\n`。特殊結(jié)束符方式的問題是過于簡(jiǎn)單的思考了協(xié)議傳輸?shù)倪^程，對(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)容我們使用表格來展示。
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)用路徑貼出來。
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")) {
// 無 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ì)分析過 FailoverClusterInvoker，因此本節(jié)略過 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;
// 省略其他方法
}上面的代碼來自 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 {
//通過取模獲得其中一個(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) {//異步無返回值
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;
}
}
//省略無關(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è)置了編解碼器，其中通過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é)開始設(shè)置
Bytes.long2bytes(req.getId(), header, 4);
// 獲取 buffer 當(dāng)前的寫位置
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();
// 獲取寫入的字節(jié)數(shù)，也就是消息體長(zhǎng)度
int len = bos.writtenBytes();
checkPayload(channel, len);
// 將消息體長(zhǎng)度寫入到消息頭中
Bytes.int2bytes(len, header, 12);
// 將 buffer 指針移動(dòng)到 savedWriteIndex，為寫消息頭做準(zhǔn)備
buffer.writerIndex(savedWriteIndex);
// 從 savedWriteIndex 下標(biāo)處寫入消息頭
buffer.writeBytes(header);
// 設(shè)置新的 writerIndex，writerIndex = 原寫下標(biāo) + 消息頭長(zhǎng)度 + 消息體長(zhǎng)度
buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
}
// 省略其他方法
}以上就是請(qǐng)求對(duì)象的編碼過程，該過程首先會(huì)通過位運(yùn)算將消息頭寫入到 header 數(shù)組中。然后對(duì) Request 對(duì)象的 data 字段執(zhí)行序列化操作，序列化后的數(shù)據(jù)最終會(huì)存儲(chǔ)到 ChannelBuffer 中。序列化操作執(zhí)行完后，可得到數(shù)據(jù)序列化后的長(zhǎng)度 len，緊接著將 len 寫入到 header 指定位置處。最后再將消息頭字節(jié)數(shù)組 header 寫入到 ChannelBuffer 中，整個(gè)編碼過程就結(jié)束了。本節(jié)的最后，我們?cè)賮砜匆幌?Request 對(duì)象的 data 字段序列化過程，也就是 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)求的過程就分析完了，接下來我們來看一下服務(wù)提供方是如何接收請(qǐng)求的。
提供方接收請(qǐng)求
(1) 請(qǐng)求解碼
這里直接分析請(qǐng)求數(shù)據(jù)的解碼邏輯，忽略中間過程，如下：
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;
}
}
// 通過 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);
}
}
}
}
}上面方法通過檢測(cè)消息頭中的魔數(shù)是否與規(guī)定的魔數(shù)相等，提前攔截掉非常規(guī)數(shù)據(jù)包，比如通過 telnet 命令行發(fā)出的數(shù)據(jù)包。接著再對(duì)消息體長(zhǎng)度，以及可讀字節(jié)數(shù)進(jìn)行檢測(cè)。最后調(diào)用 decodeBody 方法進(jìn)行后續(xù)的解碼工作，ExchangeCodec 中實(shí)現(xiàn)了 decodeBody 方法，但因其子類 DubboCodec 覆寫了該方法，所以在運(yùn)行時(shí) DubboCodec 中的 decodeBody 方法會(huì)被調(diào)用。下面我們來看一下該方法的代碼。
public class DubboCodec extends ExchangeCodec implements Codec2 {
@Override
protected Object decodeBody(Channel channel, InputStream is, byte[] header) throws IOException {
// 獲取消息頭中的第三個(gè)字節(jié)，并通過邏輯與運(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);
// 通過邏輯與運(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());
// 通過邏輯與運(yùn)算得到通信方式，并設(shè)置到 Request 對(duì)象中
req.setTwoWay((flag & FLAG_TWOWAY) != 0);
// 通過位運(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ù)解析出來
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) {
// 若解碼過程中出現(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ù)解析出來。
(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 ...");
}
}
// 其他消息類型通過 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());
// 通過 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)出過程中會(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();
}
// 忽略其他方法
}在之前課程中介紹過，服務(wù)全部暴露完成之后保存到exporterMap中。這里就是通過serviceKey獲取exporter之后獲取Invoker，并通過 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í)通過 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í)通過 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)用過程就分析完了。最后把調(diào)用過程貼出來，如下：
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ù)提供方也是通過 NettyChannel 的 send 方法將 Response 對(duì)象返回，這里就不在重復(fù)分析了。本節(jié)我們僅需關(guān)注 Response 對(duì)象的編碼過程即可。
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();
// 獲取寫入的字節(jié)數(shù)，也就是消息體長(zhǎng)度
int len = bos.writtenBytes();
checkPayload(channel, len);
// 將消息體長(zhǎng)度寫入到消息頭中
Bytes.int2bytes(len, header, 12);
// 將 buffer 指針移動(dòng)到 savedWriteIndex，為寫消息頭做準(zhǔn)備
buffer.writerIndex(savedWriteIndex);
// 從 savedWriteIndex 下標(biāo)處寫入消息頭
buffer.writeBytes(header);
// 設(shè)置新的 writerIndex，writerIndex = 原寫下標(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ì)象編碼的過程，和前面分析的 Request 對(duì)象編碼過程很相似。如果大家能看 Request 對(duì)象的編碼邏輯，那么這里的 Response 對(duì)象的編碼邏輯也不難理解，就不多說了。接下來我們?cè)賮矸治鲭p向通信的最后一環(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。接下來 NettyHandler 會(huì)將這個(gè)對(duì)象繼續(xù)向下傳遞，最后 AllChannelHandler 的 received 方法會(huì)收到這個(gè)對(duì)象，并將這個(gè)對(duì)象派發(fā)到線程池中。這個(gè)過程和服務(wù)提供方接收請(qǐng)求的過程是一樣的，因此這里就不重復(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)用過程正常
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) {
// 解碼過程中出現(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)用過程出現(xiàn)了異常
else {
// 反序列化異常信息，并設(shè)置到 Response 對(duì)象中
res.setErrorMessage(deserialize(s, channel.getUrl(), is).readUTF());
}
return res;
} else {
// 對(duì)請(qǐng)求數(shù)據(jù)進(jìn)行解碼，前面已分析過，此處忽略
}
}
}以上就是響應(yīng)數(shù)據(jù)的解碼過程，上面邏輯看起來是不是似曾相識(shí)。對(duì)的，我們?cè)谇懊嬲鹿?jié)分析過 DubboCodec 的 decodeBody 方法中關(guān)于請(qǐng)求數(shù)據(jù)的解碼過程，該過程和響應(yīng)數(shù)據(jù)的解碼過程很相似。下面，我們繼續(xù)分析調(diào)用結(jié)果的反序列化過程
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 變量的由來)中獲取返回值類型，接著對(duì)調(diào)用結(jié)果進(jìn)行反序列化，并將序列化后的結(jié)果存儲(chǔ)起來。最后對(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()));
}
}通過`CompletableFuture#complete`方法來設(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)單說明一下。一般情況下，服務(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è)問題，如何將每個(gè)響應(yīng)對(duì)象傳遞給相應(yīng)的 Future 對(duì)象，不出錯(cuò)。答案是通過調(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è)過程大致如下圖：
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è)賮砜纯?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();
//開啟心跳失敗之后處理重連，斷連的邏輯定時(shí)任務(wù)
startReconnectTask(url);
//開啟發(fā)送心跳請(qǐng)求定時(shí)任務(wù)
startHeartBeatTask(url);
}
} Dubbo 在 `HeaderExchangeClient `初始化時(shí)開啟了兩個(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)介紹過，**Dubbo 采取的是雙向心跳設(shè)計(jì)**，即服務(wù)端會(huì)向客戶端發(fā)送心跳，客戶端也會(huì)向服務(wù)端發(fā)送心跳，接收的一方更新 lastRead 字段，發(fā)送的一方更新 lastWrite 字段，超過心跳間隙的時(shí)間，便發(fā)送心跳請(qǐng)求給對(duì)端。這里的 lastRead/lastWrite 同樣會(huì)被同一個(gè)通道上的普通調(diào)用更新，通過更新這兩個(gè)字段，實(shí)現(xiàn)了只在連接空閑時(shí)才會(huì)真正發(fā)送空閑報(bào)文的機(jī)制，符合我們一開始科普的做法。
處理重連和斷連
繼續(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ù)端類型來對(duì)連接做不同的處理，當(dāng)超過設(shè)置的心跳總時(shí)間之后，客戶端選擇的是重新連接，服務(wù)端則是選擇直接斷開連接。這樣的考慮是合理的，客戶端調(diào)用是強(qiáng)依賴可用連接的，而服務(wù)端可以等待客戶端重新建立連接。
Dubbo 對(duì)于建立的每一個(gè)連接，同時(shí)在客戶端和服務(wù)端開啟了 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ù)端則采取斷連的措施。
javaee
科列表頁面右側(cè)廣告圖.jpg)
python大數(shù)據(jù)
web
科列表頁面右側(cè)廣告圖230x148.jpg)
ui
據(jù)學(xué)科列表頁右側(cè).png)
cloud
科列表頁面右側(cè)廣告圖.jpg)
test
c
netmarket
pm
Linux
movies
robot
uids
AI
.jpg)
jdbc