程式設計師突破成長的好夥伴連線技術 接力價值、

責編 | 姜新城

 第  714篇技術好文：4350字 |11分鐘閱讀

分散式系統各個角色間的通訊效率很關鍵，通訊效率的高低直接影響系統性能，基於Socket實現一個高效的Tcp通訊協議是個很有挑戰的事情，本節說明RocketMQ是如何解決這個問題的

01

Remoting模組

_____

RocketMQ的通訊相關程式碼在Remoting模組裡，先來看看主要類結構。

圖1-1 Remoting模組的類繼承關係

RemotingService為最上層介面,定義了三個方法：

void start();

void shutdown();

void registerRPCHook(RPCHookrpcHook);

RemotingClient，RemotingServer繼承RemotingService介面, 並增加了自己特有的方法。

程式碼清單1-1RemotingClient主要函式定義

1void registerProcessor(final int requestCode, finalNettyRequestProcessor processor,final ExecutorService executor);
2RemotingCommand invokeSync(final String addr, final RemotingCommandrequest, final long timeoutMillis)
 
;
3void invokeAsync(final String addr, final RemotingCommand request,final long timeoutMillis,final InvokeCallback invokeCallback);
4void invokeOneway(final String addr, final RemotingCommand request,final long timeoutMillis);
5void updateNameServerAddressList(final List<String> addrs);

然後看看具體的實現類，

NettyRemotingClient和NettyRemotingServer分別實現了RemotingClient和RemotingServer, 而且都繼承了NettyRemotingAbstract類.

通過上面的封裝，RocketMQ各個模組間的通訊，可以通過傳送統一格式的自定義訊息(RemotingCommand)來完成的，各個模組間的通訊實現簡潔明瞭。

比如NameServer模組中，NameServerController有個remotingServer變數，NameServer在啟動時初始化好各個變數，然後啟動remotingServer即可，剩下NameServer要做的是專心實現好處理RemotingCommand的邏輯。

程式碼清單1-2NameServer處理主流程程式碼

1@Override
 2public RemotingCommand processRequest(ChannelHandlerContext ctx,RemotingCommand request) throws RemotingCommandException {
 3    if (log.isDebugEnabled()){
 4       log.debug("receive request, {} {} {}",
 5            request.getCode(),
 6           RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
 7            request);
 8    }
 9    switch (request.getCode()){
10        caseRequestCode.PUT_KV_CONFIG:
11            returnthis.putKVConfig(ctx, request);
12        caseRequestCode.GET_KV_CONFIG:
13            returnthis.getKVConfig(ctx, request);
14        caseRequestCode.DELETE_KV_CONFIG:
15            returnthis.deleteKVConfig(ctx, request);
16        caseRequestCode.REGISTER_BROKER:
17            VersionbrokerVersion = MQVersion.value2Version(request.getVersion());
18            if (brokerVersion.ordinal()>= MQVersion.Version.V3_0_11.ordinal()) {
19                returnthis.registerBrokerWithFilterServer(ctx, request);
20            } else {
21                returnthis.registerBroker(ctx, request);
22            }
23        caseRequestCode.GET_HAS_UNIT_SUB_UNUNIT_TOPIC_LIST:
24            returnthis.getHasUnitSubUnUnitTopicList(ctx, request);
25        caseRequestCode.UPDATE_NAMESRV_CONFIG:
26            returnthis.updateConfig(ctx, request);
27        caseRequestCode.GET_NAMESRV_CONFIG:
28            returnthis.getConfig(ctx, request);
29        default:
30            break;
31    }
32    return null;
33}

在Consumer的原始碼中，獲取訊息的底層的通訊部分也是傳送一個RemotingComand 請求，返回的response也是個RemotingCommand型別。

程式碼清單1-3Consumer請求訊息底層實現程式碼

1private PullResult pullMessageSync(//
2    final String addr, // 1
3    final RemotingCommandrequest, // 2
4    final long timeoutMillis//3
5) throws RemotingException, InterruptedException, MQBrokerException{
6    RemotingCommand response =this.remotingClient.invokeSync(addr, request, timeoutMillis);
7    assert response != null;
8    returnthis.processPullResponse(response);
9}

從原始碼中可以看出，RocketMQ中複雜的通訊過程，被RemotingCommand統一起來，大部分的邏輯都是通過傳送Command，接受並處理Command完成。

02

協議設計和編解碼

_____

RocketMQ自己定義了一個通訊協議，使得模組間傳輸的二進位制訊息和有意義的內容之間互相轉換。協議格式如圖1-2所示。

圖1-2 RocketMQ的通訊協議

（1）第一部分是大端4個位元組整數，值等於第二，三，四部分長度總和

（2）第二部分是大端4個位元組整數，值等於第三部分的長度

（3）第三部分是通過json 序列化的資料

（4）第四部分是通過應用自定義二進位制序列化的資料

訊息的解碼過程在RomotingCommand的decode函式裡。

程式碼清單1-4訊息解碼函式

 1public static RemotingCommand decode(final ByteBuffer byteBuffer) {
 2    int length =byteBuffer.limit();
 3    int oriHeaderLen =byteBuffer.getInt();
 4    int headerLength =getHeaderLength(oriHeaderLen);
 5    byte[] headerData = newbyte[headerLength];
 6   byteBuffer.get(headerData);
 7    RemotingCommand cmd =headerDecode(headerData, getProtocolType(oriHeaderLen));
 8    int bodyLength = length - 4 - headerLength;
 9    byte[] bodyData = null;
10    if (bodyLength > 0) {
11        bodyData = newbyte[bodyLength];
12       byteBuffer.get(bodyData);
13    }
14    cmd.body = bodyData;
15    return cmd;
16}

對應的訊息編碼過程在RemotingCommand的encode函式中。

程式碼清單1-5訊息編碼函式

 1public ByteBuffer encode() {
 2    // 1> header lengthsize
 3    int length = 4;
 4    // 2> header datalength
 5    byte[] headerData =this.headerEncode();
 6    length +=headerData.length;
 7    // 3> body data length
 8    if (this.body != null) {
 9        length += body.length;
10    }
11    ByteBuffer result =ByteBuffer.allocate(4 + length);
12    // length
13    result.putInt(length);
14    // header length
15   result.put(markProtocolType(headerData.length,serializeTypeCurrentRPC));
16    // header data
17    result.put(headerData);
18    // body data;
19    if (this.body != null) {
20        result.put(this.body);
21    }
22    result.flip();
23    return result;
24}

03

Netty庫

_____

RocketMQ是基於Netty庫來完成RemotingServer和RemotingClient具體的通訊實現的，Netty是個事件驅動的網路程式設計框架，它遮蔽了Java Socket，Nio等複雜細節，使用者只需用好Netty，就可以實現一個網路程式設計專家+併發程式設計專家水平的Server、Client網路程式。應用Netty有一定的門檻，需要了解它的EventLoopGroup，Channel，Handler模型以及各種具體的配置。RocketMQ利用Netty實現的通訊類是NettyRemotingServer和NettyRemotingClient，使用者也可以參考這兩個類的實現來學習使用Netty。