大家好，我是華仔, 又跟大家見面了。

上篇主要帶大家深度剖析了「發(fā)送網(wǎng)絡(luò) I/O 的 Sender 線程的架構(gòu)設(shè)計(jì)」，消息先被暫存然后調(diào)用網(wǎng)絡(luò)I/O組件進(jìn)行發(fā)送，今天主要聊聊「真正進(jìn)行網(wǎng)絡(luò) I/O 的 NetworkClient 的架構(gòu)設(shè)計(jì)」深度剖析下消息是如何被發(fā)送出去的。

認(rèn)真讀完這篇文章，我相信你會(huì)對 Kafka NetworkClient 的源碼有更加深刻的理解。

(相關(guān)資料圖)

這篇文章干貨很多，希望你可以耐心讀完。一、總的概述

繼續(xù)通過「場景驅(qū)動(dòng)」的方式，來看看消息是如何在客戶端被累加和待發(fā)送的。

在上篇中，我們知道了消息被 Sender 子線程先暫存到 KafkaChannel 的 Send 字段中，然后調(diào)用 NetworkClient#client.poll() 進(jìn)行真正發(fā)送出去，如下圖所示「6-11步」。

NetworkClient 為「生產(chǎn)者」、「消費(fèi)者」、「服務(wù)端」等上層業(yè)務(wù)提供了網(wǎng)絡(luò)I/O的能力。在 NetworkClient 內(nèi)部使用了前面介紹的 Kafka 對 NIO 的封裝組件，同時(shí)做了一定的封裝，最終實(shí)現(xiàn)了網(wǎng)絡(luò)I/O能力。NetworkClient 不僅僅用于客戶端與服務(wù)端的通信，也用于服務(wù)端之間的通信。

接下來我們就來看看，「NetworkClient 網(wǎng)絡(luò)I/O組件的架構(gòu)實(shí)現(xiàn)以及發(fā)送處理流程」，為了方便大家理解，所有的源碼只保留骨干。

二、NetworkClient 架構(gòu)設(shè)計(jì)

NetworkClient 類是 KafkaClient 接口的實(shí)現(xiàn)類，它內(nèi)部的重要字段有「Selectable」、「InflightRequest」以及內(nèi)部類 「MetadataUpdate」。

github 源碼地址如下：

https://github.com/apache/kafka/blob/2.7.0/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java

https://github.com/apache/kafka/blob/2.7.0/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java

https://github.com/apache/kafka/blob/2.7.0/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java

https://github.com/apache/kafka/blob/2.7.0/clients/src/main/java/org/apache/kafka/clients/ClientRequest.java

https://github.com/apache/kafka/blob/2.7.0/clients/src/main/java/org/apache/kafka/clients/ClientResponse.java

1、關(guān)鍵字段

public class NetworkClient implements KafkaClient {    // 狀態(tài)枚舉值    private enum State {        ACTIVE,        CLOSING,        CLOSED    }    /* the selector used to perform network i/o */    // 用于執(zhí)行網(wǎng)絡(luò) I/O 的選擇器    private final Selectable selector;    // Metadata元信息的更新器, 它可以嘗試更新元信息    private final MetadataUpdater metadataUpdater;    /* the state of each node"s connection */    // 管理集群所有節(jié)點(diǎn)連接的狀態(tài)    private final ClusterConnectionStates connectionStates;    /* the set of requests currently being sent or awaiting a response */    // 當(dāng)前正在發(fā)送或等待響應(yīng)的請求集合    private final InFlightRequests inFlightRequests;    /* the socket send buffer size in bytes */    // 套接字發(fā)送數(shù)據(jù)的緩沖區(qū)的大?。ㄒ宰止?jié)為單位）    private final int socketSendBuffer;    /* the socket receive size buffer in bytes */    // 套接字接收數(shù)據(jù)的緩沖區(qū)的大小（以字節(jié)為單位）    private final int socketReceiveBuffer;    /* the client id used to identify this client in requests to the server */    // 表示客戶端id，標(biāo)識(shí)客戶端身份    private final String clientId;    /* the current correlation id to use when sending requests to servers */    // 向服務(wù)器發(fā)送請求時(shí)使用的當(dāng)前關(guān)聯(lián) ID    private int correlation;    /* default timeout for individual requests to await acknowledgement from servers */    // 單個(gè)請求等待服務(wù)器確認(rèn)的默認(rèn)超時(shí)    private final int defaultRequestTimeoutMs;    /* time in ms to wait before retrying to create connection to a server */    // 重連的退避時(shí)間    private final long reconnectBackoffMs;    /**     * True if we should send an ApiVersionRequest when first connecting to a broker.     * 是否需要與 Broker 端的版本協(xié)調(diào)，默認(rèn)為 true     * 如果為 true 當(dāng)?shù)谝淮芜B接到一個(gè) broker 時(shí)，應(yīng)當(dāng)發(fā)送一個(gè) version 的請求，用來得知 broker 的版本， 如果為 false 則不需要發(fā)送 version 的請求。     */    private final boolean discoverBrokerVersions;    // broker 端版本    private final ApiVersions apiVersions;    // 存儲(chǔ)著要發(fā)送的版本請求，key 為 nodeId，value 為構(gòu)建請求的 Builder    private final Map nodesNeedingApiVersionsFetch = new HashMap<>();    // 取消的請求集合    private final List abortedSends = new LinkedList<>();

從該類屬性字段來看比較多，這里說幾個(gè)關(guān)鍵字段：

selector：Kafka 自己封裝的 Selector，該選擇器負(fù)責(zé)監(jiān)聽「網(wǎng)絡(luò)I/O事件」、「網(wǎng)絡(luò)連接」、「讀寫操作」。metadataUpdater：NetworkClient 的內(nèi)部類，主要用來實(shí)現(xiàn)Metadata元信息的更新器, 它可以嘗試更新元信息。connectionStates：管理集群所有節(jié)點(diǎn)連接的狀態(tài)，底層使用 Map實(shí)現(xiàn)，NodeConnectionState 枚舉值表示連接狀態(tài)，并且記錄了最后一次連接的時(shí)間戳。inFlightRequests：用來保存當(dāng)前正在發(fā)送或等待響應(yīng)的請求集合。socketSenderBuffer：表示套接字發(fā)送數(shù)據(jù)的緩沖區(qū)的大小。socketReceiveBuffer：表示套接字接收數(shù)據(jù)的緩沖區(qū)的大小。clientId：表示客戶端id，標(biāo)識(shí)客戶端身份。reconnectBackoffMs：表示重連的退避事件，為了防止短時(shí)間內(nèi)大量重連造成的網(wǎng)絡(luò)壓力，設(shè)計(jì)了這么一個(gè)時(shí)間段，在此時(shí)間段內(nèi)不得重連。

2、關(guān)鍵方法

NetworkClient 類的方法也不少，這里針對關(guān)鍵方法逐一講解下。

（1）ready()

/** * Begin connecting to the given node, return true if we are already connected and ready to send to that node. * * @param node The node to check * @param now The current timestamp * @return True if we are ready to send to the given node */ @Override public boolean ready(Node node, long now) {     // 空節(jié)點(diǎn)     if (node.isEmpty())        throw new IllegalArgumentException("Cannot connect to empty node " + node);     // 1、判斷節(jié)點(diǎn)是否準(zhǔn)備好發(fā)送請求     if (isReady(node, now))        return true;     // 2、判斷節(jié)點(diǎn)連接狀態(tài)     if (connectionStates.canConnect(node.idString(), now))        // if we are interested in sending to a node and we don"t have a connection to it, initiate one        // 3、初始化連接，但此時(shí)不一定連接成功了        initiateConnect(node, now);     return false;}/** * Check if the node with the given id is ready to send more requests. * @param node The node * @param now The current time in ms * @return true if the node is ready*/@Overridepublic boolean isReady(Node node, long now) {    // if we need to update our metadata now declare all requests unready to make metadata requests first priority    // 當(dāng)發(fā)現(xiàn)正在更新元數(shù)據(jù)時(shí)，會(huì)禁止發(fā)送請求 && 當(dāng)連接沒有創(chuàng)建完畢或者當(dāng)前發(fā)送的請求過多時(shí)，也會(huì)禁止發(fā)送請求    return !metadataUpdater.isUpdateDue(now) && canSendRequest(node.idString(), now);}/** * Are we connected and ready and able to send more requests to the given connection? * 檢測連接狀態(tài)、發(fā)送請求是否過多 * @param node The node * @param now the current timestamp */ private boolean canSendRequest(String node, long now) {    // 三個(gè)條件必須都滿足    return connectionStates.isReady(node, now) && selector.isChannelReady(node) &&            inFlightRequests.canSendMore(node);}

該方法表示某個(gè)節(jié)點(diǎn)是否準(zhǔn)備好并可以發(fā)送請求，主要做了三件事：

先判斷節(jié)點(diǎn)是否已經(jīng)準(zhǔn)備好連接并接收請求了，需要滿足以下四個(gè)條件：!metadataUpdater.isUpdateDue(now)：不能是正在更新元數(shù)據(jù)的狀態(tài)，且元數(shù)據(jù)不能過期。canSendRequest(node.idString(), now)：此處有3個(gè)條件。(1)、客戶端和 node 連接是否處于 ready 狀態(tài)；(2)、客戶端和 node 的 channel 是否建立好；(3)、inFlightRequests 中對應(yīng)的節(jié)點(diǎn)是否可以接收更多的請求。如果連接好返回 true 表示準(zhǔn)備好，如果沒有準(zhǔn)備好接收請求，則會(huì)嘗試與對應(yīng)的 Node 連接，此處也需要滿足兩個(gè)條件：

首先連接必須是 isDisconnected，不能是 connecteding 狀態(tài)，即客戶端與服務(wù)端的連接狀態(tài)是沒有連接上。

兩次重試之間時(shí)間差要大于重試退避時(shí)間，目的就是為了避免網(wǎng)絡(luò)擁塞，防止重連過于頻繁造成網(wǎng)絡(luò)壓力過大。

最后初始化連接。

（2）initiateConnect()

/**  * 創(chuàng)建連接   * Initiate a connection to the given node  * @param node the node to connect to  * @param now current time in epoch milliseconds  */ private void initiateConnect(Node node, long now) {    String nodeConnectionId = node.idString();    try {        // 1、更新連接狀態(tài)為正在連接        connectionStates.connecting(nodeConnectionId, now, node.host(), clientDnsLookup);        // 獲取連接地址        InetAddress address = connectionStates.currentAddress(nodeConnectionId);        log.debug("Initiating connection to node {} using address {}", node, address);        // 2、調(diào)用 selector 嘗試異步進(jìn)行連接，后續(xù)通過selector.poll進(jìn)行監(jiān)聽事件就緒         selector.connect(nodeConnectionId,                    new InetSocketAddress(address, node.port()),                    this.socketSendBuffer,                    this.socketReceiveBuffer);    } catch (IOException e) {        log.warn("Error connecting to node {}", node, e);        // Attempt failed, we"ll try again after the backoff        connectionStates.disconnected(nodeConnectionId, now);        // Notify metadata updater of the connection failure        metadataUpdater.handleServerDisconnect(now, nodeConnectionId, Optional.empty());    }}

該方法主要是進(jìn)行初始化連接，做了兩件事：

調(diào)用 connectionStates.connecting() 更新連接狀態(tài)為正在連接。調(diào)用 selector.connect() 異步發(fā)起連接，此時(shí)不一定連接上了，后續(xù) Selector.poll() 會(huì)監(jiān)聽連接是否準(zhǔn)備好并完成連接，如果連接成功，則會(huì)將 ConnectionState 設(shè)置為 CONNECTED。

當(dāng)連接準(zhǔn)備好后，接下來我們來看下發(fā)送相關(guān)的方法。

（3）send()、doSend()

/** * ClientRequest 是客戶端的請求，封裝了 requestBuilder  */public final class ClientRequest {    // 節(jié)點(diǎn)地址    private final String destination;    // ClientRequest 中通過 requestBuilder 給不同類型的請求設(shè)置不同的請求內(nèi)容    private final AbstractRequest.Builder requestBuilder;    // 請求頭的 correlationId    private final int correlationId;    // 請求頭的 clientid    private final String clientId;    // 創(chuàng)建時(shí)間    private final long createdTimeMs;    // 是否需要進(jìn)行響應(yīng)    private final boolean expectResponse;    // 請求的超時(shí)時(shí)間    private final int requestTimeoutMs;    // 回調(diào)函數(shù) 用來處理響應(yīng)    private final RequestCompletionHandler callback;    ......}/** * Queue up the given request for sending. Requests can only be sent out to ready nodes. * @param request The request * @param now The current timestamp * 發(fā)送請求，這個(gè)方法 生產(chǎn)者和消費(fèi)者都會(huì)調(diào)用，其中 ClientRequest 表示客戶端的請求。 */ @Override public void send(ClientRequest request, long now) {     doSend(request, false, now); }  // 檢測請求版本是否支持，如果支持則發(fā)送請求 private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long now) {        // 確認(rèn)是否活躍        ensureActive();        // 目標(biāo)節(jié)點(diǎn)id        String nodeId = clientRequest.destination();        // 是否是 NetworkClient 內(nèi)部請求 這里為 false        if (!isInternalRequest) {             // 檢測是否可以向指定 Node 發(fā)送請求，如果還不能發(fā)送請求則拋異常             if (!canSendRequest(nodeId, now))                throw new IllegalStateException("Attempt to send a request to node " + nodeId + " which is not ready.");        }        AbstractRequest.Builder builder = clientRequest.requestBuilder();        try {            // 檢測版本            NodeApiVersions versionInfo = apiVersions.get(nodeId);            // ... 忽略            // builder.build()是 ProduceRequest.Builder，結(jié)果是ProduceRequest            // 調(diào)用 doSend 方法            doSend(clientRequest, isInternalRequest, now, builder.build(version));        } catch (UnsupportedVersionException unsupportedVersionException) {            log.debug("Version mismatch when attempting to send {} with correlation id {} to {}", builder, clientRequest.correlationId(), clientRequest.destination(), unsupportedVersionException);           // 請求的版本不協(xié)調(diào)，那么生成 clientResponse           ClientResponse clientResponse = new ClientResponse(clientRequest.makeHeader(builder.latestAllowedVersion()),                    clientRequest.callback(), clientRequest.destination(), now, now,                    false, unsupportedVersionException, null, null);            // 添加到 abortedSends 集合里            abortedSends.add(clientResponse);        }  }  /**   * isInternalRequest 表示發(fā)送前是否需要驗(yàn)證連接狀態(tài)，如果為 true 則表示客戶端已經(jīng)確定連接是好的   * request表示請求體   */  private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long now, AbstractRequest request) {        // 目標(biāo)節(jié)點(diǎn)地址        String destination = clientRequest.destination();        // 生成請求頭        RequestHeader header = clientRequest.makeHeader(request.version());        if (log.isDebugEnabled()) {            log.debug("Sending {} request with header {} and timeout {} to node {}: {}",                clientRequest.apiKey(), header, clientRequest.requestTimeoutMs(), destination, request);        }        // 1、構(gòu)建 NetworkSend 對象 結(jié)合請求頭和請求體，序列化數(shù)據(jù)，保存到 NetworkSend         Send send = request.toSend(destination, header);        // 2、構(gòu)建 inFlightRequest 對象 保存了發(fā)送前的所有信息        InFlightRequest inFlightRequest = new InFlightRequest(                clientRequest,                header,                isInternalRequest,                request,                send,                now);        // 3、把 inFlightRequest 加入 inFlightRequests 集合里        this.inFlightRequests.add(inFlightRequest);        // 4、調(diào)用 Selector 異步發(fā)送數(shù)據(jù)，并將 send 和對應(yīng) kafkaChannel 綁定起來，并開啟該 kafkaChannel 底層 socket 的寫事件，等待下一步真正的網(wǎng)絡(luò)發(fā)送        selector.send(send);}@Overridepublic boolean active() {    // 判斷狀態(tài)是否是活躍的    return state.get() == State.ACTIVE;}// 確認(rèn)是否活躍private void ensureActive() {   if (!active())      throw new DisconnectException("NetworkClient is no longer active, state is " + state);}

從上面源碼可以看出此處發(fā)送并不是真正的網(wǎng)絡(luò)發(fā)送，而是先將數(shù)據(jù)發(fā)送到緩存中。

首先最外層是 send() ，里面調(diào)用 doSend() 。這里的 doSend() 主要的作用是判斷 inFlightRequests 集合上對應(yīng)的節(jié)點(diǎn)是不是能發(fā)送請求，需要滿足三個(gè)條件：客戶端和 node 連接是否處于 ready 狀態(tài)。客戶端和 node 的 channel 是否建立好。inFlightRequests 集合中對應(yīng)的節(jié)點(diǎn)是否可以接收更多的請求。最后再次調(diào)用另一個(gè) doSend()，用來最終的請求發(fā)送到緩存中。步驟如下：構(gòu)建 NetworkSend 對象 結(jié)合請求頭和請求體，序列化數(shù)據(jù)，保存到 NetworkSend。構(gòu)建 inFlightRequest 對象。把 inFlightRequest 加入 inFlightRequests 集合里等待響應(yīng)。調(diào)用Selector異步發(fā)送數(shù)據(jù)，并將 send 和對應(yīng) kafkaChannel 綁定起來，并開啟該 kafkaChannel 底層 socket 的寫事件，等待下一步真正的網(wǎng)絡(luò)發(fā)送。

綜上可以得出這里的發(fā)送過程其實(shí)是把要發(fā)送的請求先封裝成 inFlightRequest，然后放到 inFlightRequests 集合里，然后放到對應(yīng) channel 的字段 NetworkSend 里緩存起來。總之，這里的發(fā)送過程就是為了下一步真正的網(wǎng)絡(luò)I/O發(fā)送而服務(wù)的。

接下來看下真正網(wǎng)絡(luò)發(fā)送的方法。

（4）poll()

該方法執(zhí)行網(wǎng)絡(luò)發(fā)送并把響應(yīng)結(jié)果「pollSelectionKeys 的各種讀寫」做各種狀態(tài)處理，此處是通過調(diào)用 handleXXX() 方法進(jìn)行處理的，代碼如下：

/** * Do actual reads and writes to sockets. * @param timeout The maximum amount of time to wait (in ms) for responses if there are none immediately, * must be non-negative. The actual timeout will be the minimum of timeout, request timeout and * metadata timeout * @param now The current time in milliseconds * @return The list of responses received*/@Overridepublic List poll(long timeout, long now) {   // 確認(rèn)是否活躍   ensureActive();   // 取消發(fā)送是否為空   if (!abortedSends.isEmpty()) {      // If there are aborted sends because of unsupported version exceptions or disconnects,      // handle them immediately without waiting for Selector#poll.      List responses = new ArrayList<>();      handleAbortedSends(responses);      completeResponses(responses);      return responses;   }   // 1、嘗試更新元數(shù)據(jù)   long metadataTimeout = metadataUpdater.maybeUpdate(now);   try {      // 2、執(zhí)行網(wǎng)絡(luò) I/O 操作，真正讀寫發(fā)送的地方，如果客戶端的請求被完整的處理過了，會(huì)加入到completeSends 或 complteReceives 集合中      this.selector.poll(Utils.min(timeout, metadataTimeout, defaultRequestTimeoutMs));   } catch (IOException e) {      log.error("Unexpected error during I/O", e);   }   // process completed actions   long updatedNow = this.time.milliseconds();   // 響應(yīng)結(jié)果集合：真正的讀寫操作, 會(huì)生成responses   List responses = new ArrayList<>();   // 3、完成發(fā)送的handler，處理 completedSends 集合   handleCompletedSends(responses, updatedNow);   // 4、完成接收的handler，處理 completedReceives 隊(duì)列   handleCompletedReceives(responses, updatedNow);   // 5、斷開連接的handler，處理 disconnected 列表   handleDisconnections(responses, updatedNow);   // 6、處理連接的handler，處理 connected 列表   handleConnections();   // 7、處理版本協(xié)調(diào)請求（獲取api版本號(hào)） handler   handleInitiateApiVersionRequests(updatedNow);   // 8、超時(shí)連接的handler，處理超時(shí)連接集合   handleTimedOutConnections(responses, updatedNow);   // 9、超時(shí)請求的handler，處理超時(shí)請求集合   handleTimedOutRequests(responses, updatedNow);   // 10、完成響應(yīng)回調(diào)   completeResponses(responses);   return responses;}

這里的步驟比較多，我們按照先后順序講解下。

嘗試更新元數(shù)據(jù)。調(diào)用 Selector.poll() 執(zhí)行真正網(wǎng)絡(luò) I/O 操作，可以點(diǎn)擊查看 圖解 Kafka 源碼網(wǎng)絡(luò)層實(shí)現(xiàn)機(jī)制之 Selector 多路復(fù)用器 主要操作以下3個(gè)集合。connected集合：已經(jīng)完成連接的 Node 節(jié)點(diǎn)集合。completedReceives集合：接收完成的集合，即 KafkaChannel 上的 NetworkReceive 寫滿后會(huì)放入這個(gè)集合里。completedSends集合：發(fā)送完成的集合，即 channel 上的 NetworkSend 讀完后會(huì)放入這個(gè)集合里。調(diào)用 handleCompletedSends() 處理 completedSends 集合。調(diào)用 handleCompletedReceives() 處理 completedReceives 隊(duì)列。調(diào)用 handleDisconnections() 處理與 Node 斷開連接的請求。調(diào)用 handleConnections() 處理 connected 列表。調(diào)用 handleInitiateApiVersionRequests() 處理版本號(hào)請求。調(diào)用 handleTimedOutConnections() 處理連接超時(shí)的 Node 集合。調(diào)用 handleTimedOutRequests() 處理 inFlightRequests 集合中的超時(shí)請求，并修改其狀態(tài)。調(diào)用 completeResponses() 完成每個(gè)消息自定義的響應(yīng)回調(diào)。

接下來看下第 3~9 步驟的方法實(shí)現(xiàn)。

（5）handleCompletedSends()

當(dāng) NetworkClient 發(fā)送完請求后，就會(huì)調(diào)用handleCompletedSends方法，表示請求已經(jīng)發(fā)送到 Broker 端了。

/** * Handle any completed request send. In particular if no response is expected consider the request complete. * @param responses The list of responses to update * @param now The current time*/private void handleCompletedSends(List responses, long now) {   // if no response is expected then when the send is completed, return it   // 1、遍歷 completedSends 發(fā)送完成的請求集合，通過調(diào)用 Selector 獲取從上一次 poll 開始的請求   for (Send send : this.selector.completedSends()) {       // 2、從 inFlightRequests 集合獲取該 Send 關(guān)聯(lián)對應(yīng) Node 的隊(duì)列取出最新的請求，但并沒有從隊(duì)列中刪除，取出后判斷這個(gè)請求是否期望得到響應(yīng)       InFlightRequest request = this.inFlightRequests.lastSent(send.destination());       // 3、是否需要響應(yīng), 如果不需要響應(yīng),當(dāng)Send請求完成時(shí),就直接返回.還是有request.completed生成的ClientResponse對象       if (!request.expectResponse) {           // 4、如果不需要響應(yīng)就取出 inFlightRequests 中該 Sender 關(guān)聯(lián)對應(yīng) Node 的 inFlightRequest，即提取最新的請求           this.inFlightRequests.completeLastSent(send.destination());           // 5、調(diào)用 completed() 生成 ClientResponse,第一個(gè)參數(shù)為null,表示沒有響應(yīng)內(nèi)容,把請求添加到 Responses 集合           responses.add(request.completed(null, now));       }   }}

該方法主要用來在客戶端發(fā)送請求后，對響應(yīng)結(jié)果進(jìn)行處理，做了五件事：

遍歷 seletor 中的 completedSends 集合，逐個(gè)處理完成的 Send 對象。從 inFlightRequests 集合獲取該 Send 關(guān)聯(lián)對應(yīng) Node 的隊(duì)列中第一個(gè)元素，但并沒有從隊(duì)列中刪除，取出后判斷這個(gè)請求是否期望得到響應(yīng)。判斷是否需要響應(yīng)。如果不需要響應(yīng)就刪除 inFlightRequests 中該 Sender 關(guān)聯(lián)對應(yīng) Node 的 inFlightRequest，對于 Kafka 來說，有些請求是不需要響應(yīng)的，對于發(fā)送完不用考慮是否發(fā)送成功的話，就構(gòu)建 callback 為 null 的 Response 對象。通過 InFlightRequest.completed()，生成 ClientResponse，第一個(gè)參數(shù)為 null 表示沒有響應(yīng)內(nèi)容，最后把 ClientResponse 添加到 Responses 集合。

從上面源碼可以看出，「completedSends」集合與「InflightRequests」集合協(xié)作的關(guān)系。

但是這里有個(gè)問題：如何保證從 Selector 返回的請求，就是對應(yīng)到 InflightRequests 集合隊(duì)列的最新的請求呢？

completedSends集合保存的是最近一次調(diào)用 poll() 方法中發(fā)送成功的請求「發(fā)送成功但還沒有收到響應(yīng)的請求集合」。而InflightRequests集合存儲(chǔ)的是已經(jīng)發(fā)送但還沒收到響應(yīng)的請求。每個(gè)請求發(fā)送都需要等待前面的請求發(fā)送完成，這樣就能保證同一時(shí)間只有一個(gè)請求正在發(fā)送，因?yàn)?Selector 返回的請求是從上一次 poll 開始的，這樣就對上了。

「completedSends」的元素對應(yīng)著「InflightRequests」集合里對應(yīng)隊(duì)列的最后一個(gè)元素， 如下圖所示：

（6）handleCompletedReceives()

當(dāng) NetworkClient 收到響應(yīng)時(shí)，就會(huì)調(diào)用handleCompletedReceives方法。

/** * Handle any completed receives and update the response list with the responses received. * @param responses The list of responses to update * @param now The current time * 處理 CompletedReceives 隊(duì)列,根據(jù)返回的響應(yīng)信息實(shí)例化 ClientResponse ,并加到響應(yīng)集合里*/private void handleCompletedReceives(List responses, long now) {   // 1、遍歷 CompletedReceives 響應(yīng)集合，通過 Selector 返回未處理的響應(yīng)   for (NetworkReceive receive : this.selector.completedReceives()) {       // 2、獲取發(fā)送請求的 Node id       String source = receive.source();       // 3、從 inFlightRequests 集合隊(duì)列獲取已發(fā)送請求「最老的請求」并刪除（從 inFlightRequests 刪除，因?yàn)閕nFlightRequests 存儲(chǔ)的是未收到請求響應(yīng)的 ClientRequest，現(xiàn)在請求已經(jīng)有響應(yīng)了，就不需要保存了）       InFlightRequest req = inFlightRequests.completeNext(source);       // 4、解析響應(yīng)，并且驗(yàn)證響應(yīng)頭，生成 responseStruct 實(shí)例       Struct responseStruct = parseStructMaybeUpdateThrottleTimeMetrics(receive.payload(), req.header,throttleTimeSensor, now);       // 生成響應(yīng)體       AbstractResponse response = AbstractResponse.parseResponse(req.header.apiKey(), responseStruct, req.header.apiVersion());             ....      // If the received response includes a throttle delay, throttle the connection.      // 流控處理      maybeThrottle(response, req.header.apiVersion(), req.destination, now);      // 5、判斷返回類型      if (req.isInternalRequest && response instanceof MetadataResponse)          // 處理元數(shù)據(jù)請求響應(yīng)          metadataUpdater.handleSuccessfulResponse(req.header, now, (MetadataResponse) response);      else if (req.isInternalRequest && response instanceof ApiVersionsResponse)          // 處理版本協(xié)調(diào)響應(yīng)          handleApiVersionsResponse(responses, req, now, (ApiVersionsResponse) response);      else          // 普通發(fā)送消息的響應(yīng)，通過 InFlightRequest.completed()，生成 ClientResponse，將響應(yīng)添加到 responses 集合中          responses.add(req.completed(response, now));    }}// 解析響應(yīng)，并且驗(yàn)證響應(yīng)頭，生成 responseStruct 實(shí)例private static Struct parseStructMaybeUpdateThrottleTimeMetrics(ByteBuffer responseBuffer, RequestHeader requestHeader, Sensor throttleTimeSensor, long now) {    // 解析響應(yīng)頭    ResponseHeader responseHeader = ResponseHeader.parse(responseBuffer,            requestHeader.apiKey().responseHeaderVersion(requestHeader.apiVersion()));    // 解析響應(yīng)體    Struct responseBody = requestHeader.apiKey().parseResponse(requestHeader.apiVersion(), responseBuffer);    // 驗(yàn)證請求頭與響應(yīng)頭的 correlation id 必須相等    correlate(requestHeader, responseHeader);    if (throttleTimeSensor != null && responseBody.hasField(CommonFields.THROTTLE_TIME_MS))            throttleTimeSensor.record(responseBody.get(CommonFields.THROTTLE_TIME_MS), now);    return responseBody;}

該方法主要用來處理接收完畢的網(wǎng)絡(luò)請求集合，做了五件事：

遍歷 selector 中的 completedReceives 集合，逐個(gè)處理完成的 Receive 對象。獲取發(fā)送請求的 Node id。從 inFlightRequests 集合隊(duì)列獲取已發(fā)送請求「最老的請求」并刪除（從 inFlightRequests 刪除，因?yàn)閕nFlightRequests 存儲(chǔ)的是未收到請求響應(yīng)的 ClientRequest，現(xiàn)在請求已經(jīng)有響應(yīng)了，就不需要保存了）。解析響應(yīng)，并且驗(yàn)證響應(yīng)頭，生成 responseStruct 實(shí)例，生成響應(yīng)體。處理響應(yīng)結(jié)果，此處分為三種情況：處理元數(shù)據(jù)請求響應(yīng)，則調(diào)用 metadataUpdater.handleSuccessfulResponse()。處理版本協(xié)調(diào)響應(yīng)，則調(diào)用 handleApiVersionsResponse()。普通發(fā)送消息的響應(yīng)，通過 InFlightRequest.completed()，生成 ClientResponse，將響應(yīng)添加到 responses 集合中。

從上面源碼可以看出，「completedReceives」集合與「InflightRequests」集合也有協(xié)作的關(guān)系,completedReceives集合指的是接收到的響應(yīng)集合，如果請求已經(jīng)收到響應(yīng)了，就可以從InflightRequests刪除了，這樣InflightRequests就起到了可以防止請求堆積的作用。

與 「completedSends」正好相反，「completedReceives」集合對應(yīng) 「InflightRequests」集合里對應(yīng)隊(duì)列的第一個(gè)元素，如下圖所示：

（7）leastLoadedNode()

/** * Choose the node with the fewest outstanding requests which is at least eligible for connection. This method will * prefer a node with an existing connection, but will potentially choose a node for which we don"t yet have a * connection if all existing connections are in use. If no connection exists, this method will prefer a node * with least recent connection attempts. This method will never choose a node for which there is no * existing connection and from which we have disconnected within the reconnect backoff period, or an active * connection which is being throttled. * * @return The node with the fewest in-flight requests. */ @Override public Node leastLoadedNode(long now) {        // 從元數(shù)據(jù)中獲取所有的節(jié)點(diǎn)        List nodes = this.metadataUpdater.fetchNodes();        if (nodes.isEmpty())            throw new IllegalStateException("There are no nodes in the Kafka cluster");        int inflight = Integer.MAX_VALUE;        Node foundConnecting = null;        Node foundCanConnect = null;        Node foundReady = null;        int offset = this.randOffset.nextInt(nodes.size());        for (int i = 0; i < nodes.size(); i++) {            int idx = (offset + i) % nodes.size();            Node node = nodes.get(idx);            // 節(jié)點(diǎn)是否可以發(fā)送請求            if (canSendRequest(node.idString(), now)) {                // 獲取節(jié)點(diǎn)的隊(duì)列大小                int currInflight = this.inFlightRequests.count(node.idString());                // 如果為 0 則返回該節(jié)點(diǎn)，負(fù)載最小                 if (currInflight == 0) {                    // if we find an established connection with no in-flight requests we can stop right away                    log.trace("Found least loaded node {} connected with no in-flight requests", node);                    return node;                } else if (currInflight < inflight) { // 如果隊(duì)列大小小于最大值                    // otherwise if this is the best we have found so far, record that                    inflight = currInflight;                    foundReady = node;                }            } else if (connectionStates.isPreparingConnection(node.idString())) {                foundConnecting = node;            } else if (canConnect(node, now)) {                if (foundCanConnect == null ||                        this.connectionStates.lastConnectAttemptMs(foundCanConnect.idString()) >                                this.connectionStates.lastConnectAttemptMs(node.idString())) {                    foundCanConnect = node;                }            } else {                log.trace("Removing node {} from least loaded node selection since it is neither ready " +                        "for sending or connecting", node);            }        }        // We prefer established connections if possible. Otherwise, we will wait for connections        // which are being established before connecting to new nodes.        if (foundReady != null) {            log.trace("Found least loaded node {} with {} inflight requests", foundReady, inflight);            return foundReady;        } else if (foundConnecting != null) {            log.trace("Found least loaded connecting node {}", foundConnecting);            return foundConnecting;        } else if (foundCanConnect != null) {            log.trace("Found least loaded node {} with no active connection", foundCanConnect);            return foundCanConnect;        } else {            log.trace("Least loaded node selection failed to find an available node");            return null;        } }

該方法主要是選出一個(gè)負(fù)載最小的節(jié)點(diǎn)，如下圖所示：

三、InflightRequests 集合設(shè)計(jì)

通過上面的代碼分析，我們知道「InflightRequests」集合的作用就是緩存已經(jīng)發(fā)送出去但還沒有收到響應(yīng)的 ClientRequest 請求集合。底層是通過ReqMap>實(shí)現(xiàn)，其中 key 是 NodeId，value 是發(fā)送到對應(yīng) Node 的 ClientRequest 請求隊(duì)列，默認(rèn)為5個(gè)，參數(shù)：max.in.flight.requests.per.connection配置請求隊(duì)列大小。它為每個(gè)連接生成一個(gè)雙端隊(duì)列，因此它能控制請求發(fā)送的速度。

其作用有以下2個(gè)：

節(jié)點(diǎn)是否正常：收集從「開始發(fā)送」到「接收響應(yīng)」這段時(shí)間的請求，來判斷要發(fā)送的 Broker 節(jié)點(diǎn)是否正常，請求和連接是否超時(shí)等等，也就是說用來監(jiān)控發(fā)送到哥哥節(jié)點(diǎn)請求是否正常。節(jié)點(diǎn)的負(fù)載情況：Deque 隊(duì)列到一定長度后就認(rèn)為某個(gè) Broker 節(jié)點(diǎn)負(fù)載過高了。

/** * The set of requests which have been sent or are being sent but haven"t yet received a response * 用來緩存已經(jīng)發(fā)送出去或者正在發(fā)送但均還沒有收到響應(yīng)的  ClientRequest 請求集合 */final class InFlightRequests {    // 每個(gè)連接最大執(zhí)行中的請求數(shù)    private final int maxInFlightRequestsPerConnection;    // 節(jié)點(diǎn) Node 至客戶端請求雙端隊(duì)列 Deque 的映射集合，key為 NodeId, value 是請求隊(duì)列    private final Map> requests = new HashMap<>();    /** Thread safe total number of in flight requests. */    // 線程安全的 inFlightRequestCount     private final AtomicInteger inFlightRequestCount = new AtomicInteger(0);    // 設(shè)置每個(gè)連接最大執(zhí)行中的請求數(shù)    public InFlightRequests(int maxInFlightRequestsPerConnection) {        this.maxInFlightRequestsPerConnection = maxInFlightRequestsPerConnection;}

這里通過「場景驅(qū)動(dòng)」的方式來講解關(guān)鍵方法，當(dāng)有新請求需要發(fā)送處理時(shí)，會(huì)在隊(duì)首入隊(duì)。而實(shí)際被處理的請求，則是從隊(duì)尾出隊(duì)，保證入隊(duì)早的請求先得到處理。

1、canSendMore()

先來看下發(fā)送條件限制， NetworkClient 調(diào)用這個(gè)方法用來判斷是否還可以向指定 Node 發(fā)送請求。

/** * Can we send more requests to this node? * @param node Node in question * @return true iff we have no requests still being sent to the given node * 判斷該連接是否還能發(fā)送請求*/public boolean canSendMore(String node) {        // 獲取節(jié)點(diǎn)對應(yīng)的雙端隊(duì)列        Deque queue = requests.get(node);        // 判斷條件 隊(duì)列為空 || (隊(duì)首已經(jīng)發(fā)送完成 && 隊(duì)列中沒有堆積更多的請求)        return queue == null || queue.isEmpty() ||               (queue.peekFirst().send.completed() && queue.size() < this.maxInFlightRequestsPerConnection);}

從上面代碼可以看出限制條件，隊(duì)列雖然可以存儲(chǔ)多個(gè)請求，但是新的請求要是加進(jìn)來?xiàng)l件是上一個(gè)請求必須發(fā)送成功。

條件判斷如下：

queue == null || queue.isEmpty()，隊(duì)列為空就能發(fā)送。判斷 queue.peekFirst().send.completed() 隊(duì)首是否發(fā)送完成。如果隊(duì)首的請求遲遲發(fā)送不出去，可能就是網(wǎng)絡(luò)的原因，因此不能繼續(xù)向此 Node 發(fā)送請求。隊(duì)首的請求與對應(yīng)的 KafkaChannel.send 字段指向的是同一個(gè)請求，為了避免未發(fā)送的消息被覆蓋掉，也不能讓 KafkaChannel.send 字段指向新請求。queue.size() < this.maxInFlightRequestsPerConnection，該條件就是為了判斷隊(duì)列中是否堆積過多請求，如果 Node 已經(jīng)堆積了很多未響應(yīng)的請求，說明這個(gè)節(jié)點(diǎn)出現(xiàn)了網(wǎng)絡(luò)擁塞，繼續(xù)再發(fā)送請求，則可能會(huì)超時(shí)。

2、add() 入隊(duì)

/** * Add the given request to the queue for the connection it was directed to * 將請求添加到隊(duì)列首部*/public void add(NetworkClient.InFlightRequest request) {        // 這個(gè)請求要發(fā)送到哪個(gè) Broker 節(jié)點(diǎn)上        String destination = request.destination;        // 從 requests 集合中根據(jù)給定請求的目標(biāo) Node 節(jié)點(diǎn)獲取對應(yīng) Deque 雙端隊(duì)列 reqs        Deque reqs = this.requests.get(destination);        // 如果雙端隊(duì)列reqs為null        if (reqs == null) {            // 構(gòu)造一個(gè)雙端隊(duì)列 ArrayDeque 類型的 reqs            reqs = new ArrayDeque<>();            // 將請求目標(biāo) Node 節(jié)點(diǎn)至 reqs 的映射關(guān)系添加到 requests 集合            this.requests.put(destination, reqs);        }        // 將請求 request 添加到 reqs 隊(duì)首        reqs.addFirst(request);        // 增加計(jì)數(shù)        inFlightRequestCount.incrementAndGet();}

3、completeNext() 出隊(duì)最老請求

/** * Get the oldest request (the one that will be completed next) for the given node * 取出該連接對應(yīng)的隊(duì)列中最老的請求 */ public NetworkClient.InFlightRequest completeNext(String node) {     // 根據(jù)給定 Node 節(jié)點(diǎn)獲取客戶端請求雙端隊(duì)列 reqs，并從隊(duì)尾出隊(duì)     NetworkClient.InFlightRequest inFlightRequest = requestQueue(node).pollLast();     // 遞減計(jì)數(shù)器     inFlightRequestCount.decrementAndGet();     return inFlightRequest; }

對比下入隊(duì)和出隊(duì)這2個(gè)方法，「入隊(duì) add()」時(shí)是通過 addFirst() 方法添加到隊(duì)首的，所以隊(duì)尾的請求是時(shí)間最久的，也是應(yīng)該先處理的，所以「出隊(duì) completeNext()」是通過 pollLast()，將隊(duì)列中時(shí)間最久的請求袁術(shù)移出進(jìn)行處理。

4、lastSent() 獲取最新請求?

/** * Get the last request we sent to the given node (but don"t remove it from the queue) * @param node The node id */ public NetworkClient.InFlightRequest lastSent(String node) {     return requestQueue(node).peekFirst(); }

5、completeLastSent() 出隊(duì)最新請求

/**  * Complete the last request that was sent to a particular node.  * @param node The node the request was sent to  * @return The request  * 取出該連接對應(yīng)的隊(duì)列中最新的請求  */ public NetworkClient.InFlightRequest completeLastSent(String node) {        // 根據(jù)給定 Node 節(jié)點(diǎn)獲取客戶端請求雙端隊(duì)列 reqs，并從隊(duì)首出隊(duì)        NetworkClient.InFlightRequest inFlightRequest = requestQueue(node).pollFirst();        // 遞減計(jì)數(shù)器        inFlightRequestCount.decrementAndGet();        return inFlightRequest; }

最后我們來看看「InflightRequests」，表示正在發(fā)送的請求，存儲(chǔ)著請求發(fā)送前的所有信息。

另外它支持生成響應(yīng) ClientResponse，當(dāng)正常收到響應(yīng)時(shí)，completed()會(huì)根據(jù)響應(yīng)內(nèi)容生成對應(yīng)的 ClientResponse，當(dāng)連接突然斷開后，disconnected() 會(huì)生成 ClientResponse 對象，代碼如下：

static class InFlightRequest {     //  請求頭     final RequestHeader header;     // 這個(gè)請求要發(fā)送到哪個(gè) Broker 節(jié)點(diǎn)上     final String destination;     // 回調(diào)函數(shù)     final RequestCompletionHandler callback;     // 是否需要進(jìn)行響應(yīng)     final boolean expectResponse;     // 請求體     final AbstractRequest request;     // 發(fā)送前是否需要驗(yàn)證連接狀態(tài)     final boolean isInternalRequest; // used to flag requests which are initiated internally by NetworkClient     // 請求的序列化數(shù)據(jù)     final Send send;     // 發(fā)送時(shí)間     final long sendTimeMs;     // 請求的創(chuàng)建時(shí)間，即 ClientRequest 的創(chuàng)建時(shí)間     final long createdTimeMs;     // 請求超時(shí)時(shí)間     final long requestTimeoutMs;     .....    /**     * 收到響應(yīng)，回調(diào)的時(shí)候據(jù)響應(yīng)內(nèi)容生成 ClientResponse     */    public ClientResponse completed(AbstractResponse response, long timeMs) {        return new ClientResponse(header, callback, destination, createdTimeMs, timeMs,                    false, null, null, response);    }        /**     * 當(dāng)連接突然斷開，也會(huì)生成 ClientResponse。     */    public ClientResponse disconnected(long timeMs, AuthenticationException authenticationException) {       return new ClientResponse(header, callback, destination, createdTimeMs, timeMs,                    true, null, authenticationException, null);    }}

中間的部分代碼請移步到星球查看。

五、完整請求流程串聯(lián)

一條完整的請求主要分為以下幾個(gè)階段：

調(diào)用 NetworkClient 的 ready()，連接服務(wù)端。調(diào)用 NetworkClient 的 poll()，處理連接。調(diào)用 NetworkClient 的 newClientRequest()，創(chuàng)建請求 ClientRequest。然后調(diào)用 NetworkClient 的 send()，發(fā)送請求。最后調(diào)用 NetworkClient 的 poll()，處理響應(yīng)。

1、創(chuàng)建連接過程

NetworkClient 發(fā)送請求之前，都需要先和 Broker 端創(chuàng)建連接。NetworkClient 負(fù)責(zé)管理與集群的所有連接。

2、生成請求過程3、發(fā)送請求過程4、處理響應(yīng)過程（1）請求發(fā)送完成（2）請求收到響應(yīng)（3）執(zhí)行處理響應(yīng)六、總結(jié)

這里，我們一起來總結(jié)一下這篇文章的重點(diǎn)。

1、開篇總述消息消息被 Sender 子線程先將消息暫存到 KafkaChannel 的 send 中，等調(diào)用「poll方法」執(zhí)行真正的網(wǎng)絡(luò)I/O 操作，從而引出了為客戶端提供網(wǎng)絡(luò) I/O 能力的 「NetworkClient 組件」。

2、帶你深度剖析了「NetworkClient 組件」 、「InflightRequests」、「ClusterConnectionState」的實(shí)現(xiàn)細(xì)節(jié)。

3、最后帶你串聯(lián)了整個(gè)消息發(fā)送請求和處理響應(yīng)的流程，讓你有個(gè)更好的整體認(rèn)知。

關(guān)鍵詞：