PartitionRequestClient

分割槽請求客戶端（PartitionRequestClient）用於發起遠端PartitionRequest請求，它也是RemoteChannel跟Netty通訊層之間進行銜接的物件。

對單一的TaskManager而言只存在一個NettyClient例項。但處於同一TaskManager中不同的任務例項可能會跟不同的遠端TaskManager上的任務之間交換資料，不同的TaskManager例項會有不同的ConnectionID（用於標識不同的IP地址）。因此，Flink採用PartitionRequestClient來對應ConnectionID，並提供了分割槽請求客戶端工廠（PartitionRequestClientFactory）來建立PartitionRequestClient並儲存ConnectionID與之的對應關係。

接下來，我們重點分析一下其請求ResultPartition的requestSubpartition方法：

public ChannelFuture requestSubpartition(
        final ResultPartitionID partitionId,
        final int subpartitionIndex,
        final RemoteInputChannel inputChannel,
        int delayMs) throws IOException {
    checkNotClosed();

    //將當前請求資料的RemoteInputChannel的例項注入到NettyClient的ChannelHandler管道的
 

    //PartitionRequestClientHandler例項中
    partitionRequestHandler.addInputChannel(inputChannel);

    //構建PartitionRequest請求物件
    final PartitionRequest request = new PartitionRequest(
            partitionId, subpartitionIndex, inputChannel.getInputChannelId());

    //構建一個ChannelFutureListener的例項，當I/O操作執行失敗後，會觸發相關的錯誤處理邏輯
 

    final ChannelFutureListener listener = new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (!future.isSuccess()) {
                partitionRequestHandler.removeInputChannel(inputChannel);
                inputChannel.onError(
                        new LocalTransportException(
                                "Sending the partition request failed.",
                                future.channel().localAddress(), future.cause()
                        ));
            }
        }
    };

    //立即傳送該請求，並註冊listener
    if (delayMs == 0) {
        ChannelFuture f = tcpChannel.writeAndFlush(request);
        f.addListener(listener);
        return f;
    }
    //如果請求需要延遲一定的時間，則延遲傳送請求
    else {
        final ChannelFuture[] f = new ChannelFuture[1];
        tcpChannel.eventLoop().schedule(new Runnable() {
            @Override
            public void run() {
                f[0] = tcpChannel.writeAndFlush(request);
                f[0].addListener(listener);
            }
        }, delayMs, TimeUnit.MILLISECONDS);

        return f[0];
    }
}

NettyMessage

Flink為基於Netty的通訊框架定義了自己的通訊訊息格式，以及相應的解編碼器。NettyMessage由固定大小的訊息頭和自定義的訊息體組成：

NettyMessage訊息頭的結構如下圖：

由上圖可見，NettyMessage訊息頭所佔的空間是固定的9個位元組。其中，Frame佔用一個整型值的空間其儲存的值為整個訊息（訊息頭和自定義的訊息體）的大小。Magic Number同樣佔用一個整形數值的空間且值是固定不變的；而Id表示訊息的型別，佔用一個位元組空間。

自定義的訊息體部分依不同的訊息型別各有不同。

服務端的訊息有：

• BufferResponse：服務端給出的Buffer響應訊息，編號為0；
• ErrorResponse：服務端的錯誤響應訊息，編號為1；

客戶端的訊息有：

• PartitionRequest：客戶端發起的分割槽請求，編號為2；
• TaskEventRequest：客戶端發起的任務事件請求，編號為3；
• CancelPartitionRequest：客戶端發起的取消分割槽請求，編號為4；
• CloseRequest：客戶端發起的關閉請求，編號為5；

另外，它定義了讀寫介面，面向的物件是Netty的位元組緩衝（ByteBuf）。解編碼器NettyMessageEncoder和NettyMessageDecoder以靜態內部類實現，分別用來在訊息的兩種表示（NettyMessage和ByteBuf）之間進行轉換。

NettyProtocol

NettyProtocol定義了基於Netty進行網路通訊時客戶端和服務端對事件的處理邏輯與順序。由於Netty中所有事件處理邏輯的程式碼都處於擴充套件自ChannelHandler介面的類中，所以，NettyProtocol約定了所有的協議實現者，必須提供服務端和客戶端處理邏輯的ChannelHandler陣列。

最終這些ChannelHandler將依據它們在陣列中的順序進行連結以形成ChannelPipeline。

PartitionRequestProtocol作為NettyProtocol唯一的實現，負責例項化並編排客戶端和服務端的ChannelHandler。按照順序連結的這些ChannelHandler可被視為“協議棧”。接下來，我們分別就客戶端和服務端的協議棧給出了圖示。

客戶端的整個的協議棧如下圖所示：

PartitionRequestProtocol構建出的客戶端協議棧將會被構建成ChannelPipeline，並註冊到客戶端引導物件Bootstrap中：

bootstrap.handler(newChannelInitializer<SocketChannel>(){
    @Override
    publicvoidinitChannel(SocketChannelchannel)throwsException{
        channel.pipeline().addLast(protocol.getClientChannelHandlers());
    }
});

服務端協議棧如下圖所示：

同客戶端協議棧，服務端協議棧也會被構建成ChannelPipeline並註冊到服務端引導物件ServerBootstrap中：

bootstrap.childHandler(newChannelInitializer<SocketChannel>(){
    @Override
    publicvoidinitChannel(SocketChannelchannel)throwsException{
        channel.pipeline().addLast(protocol.getServerChannelHandlers());
    }
});

需要注意的是，無論是客戶端還是服務端，資料都存在流入（inbound）和流出（outbound）的過程。流入對應著處理器介面為ChannelInboundHandler，而流出對應的處理器介面為ChannelOutboundHandler。因此，兩個協議方法所獲取到的ChannelHandler陣列並不是安裝其元素的絕對順序組成的管道。而是會區分其型別是流入還是流出（根據介面的型別判斷），結合不同的型別並按照其在陣列中的順序將其連結成管道。

微信掃碼關注公眾號：Apache_Flink

QQ掃碼關注QQ群：Apache Flink學習交流群（123414680）