目錄

• RocketMQ的基本通信通信流程
• RocketMQ的通信類結構
• RocketMQ的多線程模型 同步異步與阻塞非阻塞的區別 linux 網絡 I/O 模型 Netty 的多線程模型 RocketMQ 的線程模型
• 消息協議設計與編解碼
• 消息的通信方式和流程 Client 發送請求消息 Server 接收請求消息和處理邏輯
• 總結

前言

本文會對 RocketMQ 網絡通信框架進行深入的學習，讓你可以徹底掌握 RocketMQ 的網絡通信的底層原理，并且可以學習一下高性能網絡通信模塊在開源框架中是如何實現的。

RocketMQ 的基本通信流程

下圖是 RocketMQ 消息隊列的整體架構圖

NameServer

NameServer 每隔 10s 會掃描所有的 Broker 連接，如果 NameServer 發現 Broker 上一次的心跳距離當前已經超過 120s，斷開 NameServer 與 Broker 的連接，并將 Broker 相關的數據從 NameServer 中剔除掉。

Broker

Broker 啟動后會跟所有的 NameServer 建立長連接，并且每隔 30s 定時向 NameServer 上報 Topic 路由信息。

Producer

Producer 與 NameServer 集群中的其中一個節點建立長連接，默認每隔 30s 從 NameServer 中獲取所有 Topic 隊列的最新情況。假如 Broker 不可用，Producer 最多 30s 能夠感知。

Producer 與提供 topic 服務的 Broker 建立長連接，默認 30s 向所有關聯的 Broker 發送心跳，Broker 每隔 10s 掃描所有存活的連接。如果 Broker 在 120s 內沒有收到心跳數據，關閉與 Producer 建立的長連接。

Consumer

Consumer 與 NameServer 集群中的其中一個節點建立長連接，默認每隔 30s 從 NameServer 中獲取所有 Topic 隊列的最新情況。假如 Broker 不可用，Consumer 最多 30s 能夠感知。

Consumer 每隔 30s 向所有關聯的 Broker 發送心跳，Broker 每隔 10s 掃描所有存活的連接，若某個連接 120s 內沒有發送心跳數據，則關閉連接；并向該 Consumer Group 的所有 Consumer 發出通知，Group 內的 Consumer 重新分配隊列，然后繼續消費。

可以看出整個 RocketMQ 各個角色之間都存在網絡通信，因此如何設計一個良好的網絡通信模塊在 MQ 中是至關重要的，它將會決定 RocketMQ 集群整體的消息傳輸能力與最終的性能。

RocketMQ 在 rocketmq-remoting 模塊下設計了網絡通信的結構，所有用到網絡通信的模塊（nameserver、broker、客戶端）都依賴該模塊。并且為了實現客戶端與服務器之間高效的數據請求與接收，RocketMQ 消息隊列自定義了通信協議并在 Netty 的基礎上擴展了通信模塊。

RocketMQ的通信類結構

我們看一下 Remoting 通信模塊的結構類圖：

• RemotingService 是最上層的接口，定義了三個方法

void start();
void shutdown();
void registerRPCHook(RPCHook rpcHook);
復制代碼

• RemotingServer：定義了服務端的接口，繼承了上層接口 RemotingService
• RemotingClient：定義了客戶端的接口，繼承了上層 RemotingService

RemotingServer 與 RemotingClient 定義的方法是類似的，主要包含了同步、異步、oneway 方式的通信和注冊處理器 processor，其余的就是針對服務端和客戶端特定的接口方法，比如服務端根據 requestCode 獲取處理器的 getProcessorPair() 方法，客戶端獲取NameServer地址列表 getNameServerAddressList() 方法。

• NettyRemotingAbstract：Netty 通信抽象類，定義并封裝了服務端與客戶端公共方法。這個也是 RocketMQ 網絡通信的核心類。
• NettyRemotingServer：服務端的實現類，實現了RemotingServer 接口，繼承 NettyRemotingAbstract 抽象類。
• NettyRemotingClient：客戶端的實現類，實現類 RemotingClient 接口，繼承 NettyRemotingAbstract 抽象類。

RocketMQ 基于 Netty 框架的基礎上，對通信的 Server 和 Client 進行抽象和封裝的處理，使得結構更為簡潔和易擴展。

RocketMQ 基于 Netty 進行擴展，那我們來學習一下 RocketMQ 的多線程模型是如何實現的。

RocketMQ 多線程模型

學習 RocketMQ 多線程模型，需要學習下面的基本概念：

• 同步異步與阻塞非阻塞的區別
• linux 網絡 I/O 模型
• Netty 的多線程模型

同步異步與阻塞非阻塞的區別

同步和異步

關注的是結果消息的通信機制

同步：同步的意思就是調用方需要主動等待結果的返回。

異步：異步的意思就是不需要主動等待結果的返回，而是通過其它手段，比如：狀態通知，回調函數等。

阻塞與非阻塞

主要關注的是等待結果返回的調用方的狀態

阻塞：進程調用接口后，如果接口沒有準備好數據，那么這個進程會被掛起，什么都不能做，直到有數據返回的時候才會喚醒。內核將 CPU 時間片切換給其它需要的進程，那當前的進程在這種情況下就得不到 CPU 時間做該做的事情了。

非阻塞：進程調用接口，如果接口沒有準備好數據，進程也能處理后續的操作，不會被掛起，CPU 時間片不會切換給其他進程，應用程序可以得到足夠的 CPU 時間繼續完成其它事情。通過不斷的輪詢檢查數據是否已經處理完成了。由于輪詢效率太低了（每次都需要進行系統調用），就有了后面講到的 I/O 多路復用模型。

下圖是網絡編程中，socket 下面 read 與 write 方法阻塞模式與非阻塞模式的區別：

兩者的組合

同步和異步在于是否需要你去確認是否有結果。 阻塞與非阻塞在于你這期間是一直等待結果還是不等待做其它的事情。

同步阻塞

同步阻塞是編程中常見的模型，打個比方你去喜茶買杯奶茶，這個時候需要現做奶茶，你需要在店里一直等待，期間不能做任何事情，像木頭人一樣死等，等到店員做完后交給你。（不能玩手機感覺要死了）

同步非阻塞

同步非阻塞在編程模式中可以抽象為一種輪詢模式，這個時候你不需要死等了，你可以玩手機或者先出去逛一逛，然后回來問店員奶茶做好了么。

異步阻塞

異步阻塞在編程模式中用的比較少，有點像你寫了個線程池，submit 之后馬上 future.get()。這就像你買奶茶，店員再做，然后告訴你好了會叫號的，然后呢，你就瞪大眼睛一直看著號碼牌。

異步非阻塞

異步非阻塞，你不需要瞪大眼睛一直盯著號碼牌了，快玩手機吧，到了你的號碼就會通知你的。

Linux 網絡 IO 模型

CPU 處理數據的速度遠大于 IO 準備數據的速度 。所以理論上 任何編程語言 都會遇到這種 CPU 處理速度和 IO 速度不匹配的問題，在網絡編程中如何進行網絡 IO 優化，怎么高效地利用 CPU 進行網絡數據處理就變得非常重要。

首先我們要知道一個基本概念，Linux 的內核將所有外部設備都看做一個文件來操作（一切皆文件），對一個 File 的讀寫操作會調用內核提供的系統命令，返回一個 File Descriptor（FD 文件描述符）。而對一個 Socket 的讀寫也會有響應的描述符，稱為 Socket File Descriptor（Socket 文件描述符），描述符就是一個數字，指向內核中的一個結構體（文件路徑，數據區等一些屬性）。

根據 UNIX 網絡編程對 IO 模型的分類，UNIX 提供了 5 種 IO 模型。

5 種 I/O 模型主要經歷了下面兩個階段：

1. 等待數據準備好，需要從網卡將數據拷貝到內核緩沖區中
2. 將準備好的數據，從內核緩沖區中拷貝到用戶緩沖區中

下面介紹 IO 模型的時候所說的第一階段和第二階段就對應著上面的這兩個階段。

阻塞 IO 模型

read 和 write 操作都可以理解為是一個 recvfrom 的調用。

阻塞 IO 模型（Blocking IO）是最常用的 IO 模型，IO 模型兩個階段沒有準備好的情況下，所有文件操作都是阻塞的。以套接字為例：在進程空間中調用 recvfrom，其系統調用直到數據包到達且被復制到應用進程的緩沖區中或發生錯誤時才返回。在此期間一直會等待，進程從調用 recvfrom 開始到它返回的整段時間都是被阻塞的。即 recvfrom 的調用會被阻塞。

優點：程序簡單，阻塞等待數據期間進程/線程掛起，基本不會占用 CPU 資源。

缺點：每個連接需要獨立的進程/線程單獨處理，高并發的情況下，會創建大量的連接，導致內存、線程切換開銷都很大，所以不適合高并發的場景使用。

非阻塞 IO 模型

非阻塞 IO 模型（Non-blocking IO），recvfrom 從應用層到內核的時候，如果第一階段沒有準備好，緩沖區沒有數據的話，就直接返回一個 EWOULDBLOCK 錯誤，一般對非阻塞 IO 模型進行輪詢檢查這個狀態，看內核是不是有數據到來。即反復調用 recvfrom 等待成功指示（輪詢）。

優點：不會阻塞在內核等待數據的過程，也就是不會等待第一階段準備的過程，不用阻塞等待，有較好的實時性。

缺點：不斷輪詢訪問內核，每次訪問都是一次系統調用，每次系統調用都會涉及到用戶態與內核態的切換，一直占用著 CPU，系統資源利用率降低。

我們需要減少頻繁的系統調用，需要將輪詢的操作交給操作系統來進行實現，我們只需要進行一次系統調用就可以了。

IO 多路復用模型

多路復用：「多路」代表著多個網絡連接，「復用」指的是復用同一個線程。

IO 復用模型（IO Multiplexing），Linux 提供 select/poll，進程通過將一個或者多個 fd 傳遞給 select 或 poll 系統調用，阻塞在 select 操作上，阻塞在 select 操作上，這樣 select/poll 可以幫我們偵測到多個 fd 是否處于就緒狀態。select/poll 會順序掃描 fd 是否就緒，select 與 poll 區別在于 fd 數量上限制，select 限制 fd 數量為 1024，poll 對數量不限制。select/poll 系統調用有下面幾個問題：

1. select/poll 調用需要傳入 fd 數組，需要拷貝一份到內核中，高并發場景下會帶來很大的性能消耗。減少 fd 數組的復制可以減少不必要的開銷。
2. select/poll 在內核層面仍然是通過遍歷的方式檢查文件描述符的就緒狀態，是個同步的過程。如果改為異步事件通知，可以有效提高內核的工作效率。
3. select /poll 僅僅返回刻度描述文件符的個數，具體哪個可讀還需要用戶自己遍歷。如果給用戶返回就緒的文件描述符，就不需要用戶進行遍歷了。

epoll 針對以上三點進行了改進：

1. 內核會保存一份文件描述符集合，不需要用戶每次都重新傳入，只需要告訴內核修改的部分即可。
2. 內核不再通過輪詢的方式找到就緒的文件描述符，而是通過異步 I/O 事件喚醒。
3. 內核僅會將有 IO 事件的文件描述符返回給用戶，用戶也不需要遍歷整個文件描述符集合。
4. 下圖是 epoll 的流程，源自：閃客的《 你管這破玩意叫 IO 多路復用？》

信號驅動 IO 模型

首先我們允許套接口進行信號驅動 I/O, 并安裝一個信號處理函數，進程繼續運行并不阻塞。當數據準備好時，進程會收到一個 SIGIO 信號，可以在信號處理函數中調用I/O操作函數處理數據。

**優點：**線程并沒有在等待數據時被阻塞，可以提高資源的利用率。

**缺點：**信號 I/O 在大量 IO 操作時可能會因為信號隊列溢出導致沒法通知。

信號驅動 I/O 盡管對于處理 UDP 套接字來說有用，即這種信號通知意味著到達一個數據報，或者返回一個異步錯誤。

但是，對于 TCP 而言，信號驅動的 I/O 方式近乎無用，因為導致這種通知的條件為數眾多，每一個來進行判別會消耗很大資源，與前幾種方式相比優勢盡失。

異步 IO 模型

異步 IO（Asynchronous IO），告知內核啟動某個文件，并讓內核整個操作完成后（包括將數據從內核復制到用戶自己的緩沖區）通知我們。

這種模型與信號模型的主要區別是：信號驅動 IO 由內核通知我們何時可以開始一個 IO 操作：異步 IO 模型由內核通知我們 IO 操作何時已完成。

優點：異步 I/O 能夠充分利用 DMA 特性，讓 I/O 操作與計算重疊。

缺點：要實現真正的異步 I/O，操作系統需要做大量的工作。目前 Windows 下通過 IOCP 實現了真正的異步 I/O。

而在 Linux 系統下，Linux 2.6才引入，目前 AIO 并不完善，因此在 Linux 下實現高并發網絡編程時都是以 IO 復用模型模式為主。

五種 IO 模型的總結

這五種 I/O 模型中，前四種屬于同步 I/O，因為其中真正的 I/O 操作將阻塞進程/線程，只有異步 I/O 模型才與 POSIX 定義的異步 I/O 相匹配。

阻塞式I/O：簡單、高并發創建連接多，造成性能瓶頸。

非阻塞式I/O：不用阻塞等待數據，快速響應，但輪詢是在線程中進行的，會占用 CPU 時間，浪費系統資源。

I/O 多路復用：多個網絡連接可以復用同一個線程，而且多路復用的輪詢是在內核中進行，效率比線程更高。并且 epoll 比 select/poll 性能更好，直接返回發生 I/O 事件的描述符。

Netty 的多線程模型

通過上面講解，我們了解了五種 I/O 模型，Netty 是基于 NIO 實現的高性能、異步事件驅動的網絡通信框架。

基于 NIO 可以實現 I/O 多路復用模型，通過一個線程對多個連接事件進行監聽。Netty 基于此實現了 Reactor 多線程模型。

Reactor 模式：是指通過一個或者多個輸入同時傳遞給服務器的服務請求的事件驅動處理模式。

通過 I/O 多路復用統一監聽事件，收到事件后分發給線程進行處理。

單 Reactor 單線程

如圖是單 Reactor 單線程對請求的處理流程：

1. Reactor 基于 select 實現 I/O 多路復用，應用程序通過一個阻塞對象監聽多路連接請求。
2. 如果是建立連接請求事件，通過 Acceptor 處理連接請求，然后創建一個 Handler 對象處理連接完成后的后續業務處理。
3. 如果不是建立事件，則 Reactor 會分發調用連接對應的 Handler 來處理請求。
4. Handler 會完成：decode -> read -> 業務處理 -> send -> encode 完整的業務流程

這個模型是單線程處理，不會涉及上下文切換、資源競爭等問題。但是單線程處理沒有辦法發揮多核 CPU 的性能，并且 Handler 如果處理某個響應很慢，就會導致其它的請求處于長時間等待狀態。并且如果線程意外掉線或者進入死循環，會導致整個通信模塊不可用，造成節點故障。

單 Reactor 多線程

Handler 不進行業務的處理，只負責響應，業務交給 Worker 線程池進行處理。這樣可以充分利用多核 CPU 的能力，但是 Reactor 承擔所有事件監聽和響應，單線程運行，高并發場景下就很容易出現系統瓶頸。

所以需要建立連接時間與讀寫事件進行分開，就有了多 Reactor 的場景。

多 Reactor 多線程

Reactor 主線程 MainReactor 進行建立連接，然后將創建好的連接分配到 Reactor 子線程 SubReactor 進行事件的監聽。然后 SubReactor 監聽到讀寫事件，就進行請求的處理，這塊和上面的單 Reactor 多線程是一樣的了。

基于該模型，我們可以通過增加 Reactor 的實例個數來充分利用 CPU 資源，并且將各個處理進行了解耦，提高了整體的并發效率。

Netty 就是基于主從多線程模型來支持高并發網絡請求的處理。

RocketMQ 的線程模型

RocketMQ 在 Netty 原生的多線程 Reactor 模型上做了一系列的擴展和優化，如下圖所示：

其實是四個線程池，分別進行不同的處理，線程池分別對應數字：1 + N + M1 + M2。

• 一個 Reactor 主線程（eventLoopGroupBoss，即為上面的1）負責監聽 TCP 網絡連接請求，建立好連接，創建SocketChannel，并注冊到 selector 上。
• RocketMQ的源碼中會自動根據 OS 的類型選擇 NIO 和 Epoll，也可以通過參數配置, 然后監聽真正的網絡數據。（NIO 底層基于 poll 實現，Epoll 底層是基于 epoll 實現）
• 拿到網絡數據后，再丟給 Worker 線程池（eventLoopGroupSelector，即為上面的“N”，源碼中默認設置為 3 ）。
• 在真正執行業務邏輯之前需要進行 SSL 驗證、編解碼、空閑檢查、網絡連接管理，這些工作交給defaultEventExecutorGroup（即為上面的“M1”，源碼中默認設置為8）去做。
• 而處理業務操作放在業務線程池中執行，根據 RomotingCommand 的業務請求碼 code 去processorTable 這個本地緩存變量中找到對應的 processor，然后封裝成 task 任務后，提交給對應的業務 processor 處理線程池來執行（sendMessageExecutor，以發送消息為例，即為上面的 “M2”）。
• 從入口到業務邏輯的幾個步驟中線程池一直在增加，這跟每一步邏輯復雜性相關，越復雜，需要的并發通道越寬。

消息協議設計與編解碼

客戶端與服務端之間發送消息，需要對發送的消息進行一個協議約定，按照約定好的協議規則進行編碼和解碼操作。RocketMQ 自定義了 RocketMQ 的消息協議，并且為了高效地在網絡中傳輸消息和對收到消息讀取，對消息進行編解碼。

RemotingCommand 這個類在消息傳輸的過程中對所有數據內容進行封裝，包含下面的一些數據內容：

自定義的 RocketMQ 傳輸協議格式如下：

可見傳輸內容主要可以分為以下4部分：

(1) 消息長度：總長度，四個字節存儲，占用一個int類型；

(2) 序列化類型&消息頭長度：同樣占用一個int類型，第一個字節表示序列化類型，后面三個字節表示消息頭長度；

(3) 消息頭數據：經過序列化后的消息頭數據；

(4) 消息主體數據：消息主體的二進制字節數據內容；

RemotingCommand 中的編碼和解碼的源碼：

// RemotingCommand 根據傳輸協議進行編碼
public ByteBuffer encode() {
        // 1> header length size
        int length=4;

        // 2> header data length，把消息頭進行了編碼，拿到了header字節數組
        byte[] headerData=this.headerEncode();
        length +=headerData.length;

        // 3> body data length
        if (this.body !=null) {
            length +=body.length;
        }

        ByteBuffer result=ByteBuffer.allocate(4 + length);

        // 1、設置消息長度
        result.putInt(length);

        // 2、設置序列化類型+消息頭長度
        result.put(markProtocolType(headerData.length, serializeTypeCurrentRPC));

        // 3、經過序列化后的消息頭數據
        result.put(headerData);

        // 4、消息主題數據
        if (this.body !=null) {
            result.put(this.body);
        }

        result.flip();

        return result;
}
復制代碼

// RemotingCommand 根據傳輸協議進行解碼
    public static RemotingCommand decode(final byte[] array) throws RemotingCommandException {
        ByteBuffer byteBuffer=ByteBuffer.wrap(array); 
        return decode(byteBuffer);
    }

    public static RemotingCommand decode(final ByteBuffer byteBuffer) throws RemotingCommandException {
        // 解碼的過程就是編碼過程的逆向過程
        int length=byteBuffer.limit(); // 獲取總長度
        int oriHeaderLen=byteBuffer.getInt(); // 獲取消息長度
        int headerLength=getHeaderLength(oriHeaderLen);// 獲取消息頭長度

        // 搞一個頭長度的字節數組，一次性把headers都讀出來放到字節數組里去
        byte[] headerData=new byte[headerLength];
        // 獲取消息頭數據
        byteBuffer.get(headerData);

        // 對header根據協議類型進行解碼
        RemotingCommand cmd=headerDecode(headerData, getProtocolType(oriHeaderLen));

				// 獲取消息體長度
        int bodyLength=length - 4 - headerLength;
        byte[] bodyData=null;
        // 獲取消息體內容
        if (bodyLength > 0) {
            bodyData=new byte[bodyLength];
            byteBuffer.get(bodyData);
        }
        cmd.body=bodyData;

        return cmd;
    }
復制代碼

消息的通信方式和流程

RocketMQ 消息通信主要有三種方式：同步（sync），異步（async），單向（oneway）

我們就以同步通信模式為例，來分析一下客戶端發送流程和服務端接收消息并處理邏輯的流程。

Client 發送請求消息

客戶端（生產者 Producer）發送消息的時候，會直接調用 DefaultMQProducerImpl 類中的 send(Message) 進行消息的發送，默認是同步通信模式的。這個方法最底層會調用 NettyRemotingClient#invokeSync 方法，獲取到服務器與 Broker 之間的 Channel，調用 NettyRemotingAbstract#invokeSyncImpl 方法傳入 Channel 和 RemotingCommand 將消息發送給服務端（Broker）。

NettyRemotingAbstract#invokeSyncImpl 發送消息的源碼如下（已附上相關注釋）：

// 因為需要先分析出來請求，然后再進行響應的分析，分析請求肯定要根據調用 responseTable.put 的那塊作為切入點進行分析。
public RemotingCommand invokeSyncImpl(
        final Channel channel, // 網絡連接
        final RemotingCommand request, // 需要發送的請求
        final long timeoutMillis // 同步發送網絡請求的超時時間
)
        throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException {
    // requestId，每個請求的唯一ID
    final int opaque=request.getOpaque();

    try {
        // 將 Channel 和 requestID 封裝成一個 ResponseFuture
        final ResponseFuture responseFuture=new ResponseFuture(channel, opaque, timeoutMillis, null, null);
        // 將 ResponseFuture 放入到 responseTable 中，并以 requestId 作為 key。
        this.responseTable.put(opaque, responseFuture);
        final SocketAddress addr=channel.remoteAddress();
        // 使用 Netty 的 Channel 發送請求數據到服務端
        channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture f) throws Exception {
                // 請求成功，設置 ResponseFuture 的 sendRequestOK=true
                if (f.isSuccess()) {
                    responseFuture.setSendRequestOK(true);
                    return;
                } else {
                    responseFuture.setSendRequestOK(false);
                }

                responseTable.remove(opaque);
                // 請求失敗，將異常進行保存
                responseFuture.setCause(f.cause());
                responseFuture.putResponse(null);
                log.warn("send a request command to channel <" + addr + "> failed.");
            }
        });
        
        // 基于 CountDownLatch 實現同步，等待 timeoutMillis 毫秒。
        RemotingCommand responseCommand=responseFuture.waitResponse(timeoutMillis);
        // 判斷請求異常的情況
        if (null==responseCommand) {
            if (responseFuture.isSendRequestOK()) {
                throw new RemotingTimeoutException(RemotingHelper.parseSocketAddressAddr(addr), timeoutMillis,
                        responseFuture.getCause());
            } else {
                throw new RemotingSendRequestException(RemotingHelper.parseSocketAddressAddr(addr), responseFuture.getCause());
            }
        }
        return responseCommand;
    } finally {
        this.responseTable.remove(opaque);
    }
}

復制代碼

• opaque：請求唯一標識碼，客戶端會給每個請求生成一個唯一標識碼。
• ResponseFuture：獲取發送消息結果的封裝對象，內部基于 CountDownLatch 來實現消息同步通信模式。創建對象會默認創建一個 countDownLatch=new CountDownLatch(1)，調用 Channel 進行消息發送后然后調用 waitResponse(timeoutMillis)（內部是countDownLatch.await(timeout)）阻塞等待結果，請求結果到達后會通過 countDownLatch.countDown() 來釋放計數器。
• responseTable：保存 opaque 和 ResponseFuture 的映射關系，對于每個請求發送前，我們將關系保存起來，當響應到達后，可以根據唯一標識來獲取 ResponseFuture 并設置響應結果 responseCommand。

Server 接收消息和處理邏輯

Server 端接收消息的核心方法是 NettyRemotingAbstract#processRequestCommand，源碼如下（省略一些異常處理代碼）:

public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
    // 獲取請求處理組件
    final Pair<NettyRequestProcessor, ExecutorService> matched=this.processorTable.get(cmd.getCode());
    // 如果沒有獲取默認的請求處理組件
    final Pair<NettyRequestProcessor, ExecutorService> pair=null==matched ? this.defaultRequestProcessor : matched;
    // 獲取請求的id
    final int opaque=cmd.getOpaque();

    if (pair !=null) {
        Runnable run=new Runnable() {
            @Override
            public void run() {
                try {
                    String remoteAddr=RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                    doBeforeRpcHooks(remoteAddr, cmd);
                    // 設置請求回調函數
                    final RemotingResponseCallback callback=new RemotingResponseCallback() {
                        @Override
                        public void callback(RemotingCommand response) {
                            doAfterRpcHooks(remoteAddr, cmd, response);
                            if (!cmd.isOnewayRPC()) {
                                if (response !=null) {
                                    response.setOpaque(opaque);
                                    response.markResponseType();
                                    try {
                                        ctx.writeAndFlush(response);
                                    } catch (Throwable e) {
                                        log.error("process request over, but response failed", e);
                                        log.error(cmd.toString());
                                        log.error(response.toString());
                                    }
                                } else {
                                }
                            }
                        }
                    };
                    // 異步處理
                    if (pair.getObject1() instanceof AsyncNettyRequestProcessor) {
                        AsyncNettyRequestProcessor processor=(AsyncNettyRequestProcessor)pair.getObject1();
                        processor.asyncProcessRequest(ctx, cmd, callback);
                    } else {
                        // oneway 和 同步處理
                        NettyRequestProcessor processor=pair.getObject1();
                        RemotingCommand response=processor.processRequest(ctx, cmd);
                        callback.callback(response);
                    }
                } catch (Throwable e) {
                    log.error("process request exception", e);
                    log.error(cmd.toString());

                    if (!cmd.isOnewayRPC()) {
                        final RemotingCommand response=RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_ERROR,
                                RemotingHelper.exceptionSimpleDesc(e));
                        response.setOpaque(opaque);
                        ctx.writeAndFlush(response);
                    }
                }
            }
        };

        // 如果拒絕請求，就返回系統繁忙。
        if (pair.getObject1().rejectRequest()) {
            final RemotingCommand response=RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
                    "[REJECTREQUEST]system busy, start flow control for a while");
            response.setOpaque(opaque);
            ctx.writeAndFlush(response);
            return;
        }

        try {
            // 將線程和channel，requestCommand 創建一個 RequestTask
            final RequestTask requestTask=new RequestTask(run, ctx.channel(), cmd);
            // 交給執行器對應的線程池處理，對應的就是我們在 RocketMQ 多線程模型的 M2。
            pair.getObject2().submit(requestTask);
        } catch (RejectedExecutionException e) {
            // 線程池拒絕處理異常
        }
    } else {
        // 沒有處理器異常
    }
}
復制代碼

1. 根據 code 獲取處理器和線程池 Pair<NettyRequestProcessor, ExecutorService> pair 。
2. 將請求邏輯封裝成一個 Runnable，內部根據 NettyRequestProcessor 進行請求的處理，并創建一個 RemotingCommand response，將響應結果通過 Netty 調用 writeAndFlush(response) 進行返回。
3. 創建 RequestTask 封裝線程、Channel、requestCommand，交給對應的線程池進行處理，也就是 M2 線程池。

總結

本篇文章分別從多線程模型、消息協議設計與編解碼、消息通信方式這幾個方面并結合源碼深入的帶著大家了解了 RocketMQ 的整個通信模塊的設計與交互流程。并且在多線程模型中帶著大家了解阻塞非阻塞和同步異步、五種網絡I/O、Reactor 開發模式這幾個基礎概念。

關于 RocketMQ 網絡通信這塊就告一段落，后續小李會按照一個消息的發送流程，分別從 Producer 發送消息、Broker 存儲消息、Consumer 消費消息來繼續帶著大家深入剖析 RocketMQ 的底層原理。

原文鏈接：https://juejin.cn/post/7103437918366089230

2016年6月22日，由世界O2O組織（WOO）、全球移動游戲聯盟（GMGC）以及光合資本共同主辦的世界O2O博覽會暨IN+2016創新大會在北京國家會議中心盛大舉行，此次大會以“ALL IN ·ALL WIN”為主題，來自全球O2O領域各垂直細分行業的企業領袖、投資方代表、創新創業領域新秀、相關服務商代表以及國際行業組織齊聚一堂，共話年度O2O領域的熱點議題。很快商務VP黃朝斌先生以“快云，Say Hello To The World”為題發表了主題演講，以下為演講實錄:

黃朝斌：謝謝主持人，也謝謝主辦方，能夠有幸今天下午在這樣一個很好的地方和大家做一個分享，也感謝剛才張總的預告，大致也說了一下我接下來分享的內容。

首先，我做個自我介紹，我是很快的VP。很快是專注和聚焦于微信公眾號和社交媒體的第三方生態服務商，是最大的公眾號流量分發的平臺。

我今天帶來的主題是“Say Hello To The World”，當時想這個題目的時候想了很久，實際上不止一次在這樣的場合做內容分享，我突然在想，咱們是世界O2O的博覽會，第一，World這個詞是很好的印證，我們專注于微信公眾號的窗口，希望能夠通過公眾號這個窗口使客戶和外界商業的社會好好坐下來聊一聊。

我們現在來看幾個數字，第一個數字是6.88億，據可靠的官方媒體報道這是網民總數；第二個數字，16.2億，這是移動網民的總數，已經占了將近90%以上的人群，并且60%以上都是30歲以下的人。我們最大的社交平臺微信的情況，截止到去年的12月份，微信預估月活躍用戶在5.6億，這是國內的情況。在這里問大家一個事情，大家每天早上起來第一件事情做什么？我相信很多人起來第一件事情并不是起床看衣服，而是起來看看手機在哪里，朋友圈有誰的更新，大部分是這樣。微信公布了最近的用戶數據，大家可以看到很有意思，很多人早上醒來第一件事情就是刷朋友圈，上午用微信訂餐，下午發現自己家里電沒有了，用微信支付等等一系列的場景。實際上說明我們現在的微信成為了一個生態，并不僅僅是一個APP的應用，基于微信的生態，大家也可以看到，這兩年已經誕生了很多這樣的新三板的公司。

說到了微信，再說微信公眾號，公眾號將近超過一千萬，我們認為已經達到兩千萬的數字，如此龐大的公眾號誕生了很多開發的需求，我們可以看到很多線下的商家和商戶都有自己的公眾號和自己的會員系統，對于這種公眾號開發的需求是極其龐大的。熟悉公眾號的人知道，現在公眾號分三種，一種是訂閱號，一種是服務號，另外一種就是企業號，企業號更多是企業OA內部信息的交互。

微信支付的數據大概已經超過了4億，這說明我們給第三方生態的空間是非常巨大的。說到公眾號，我們現在看看公眾號對于企業的重要性，我覺得這兩年移動互聯網是非常熱的，其重要性已經無需贅言，大家工作交流的時間碎片化了，更多的是通過碎片化的時間學習和掌握你所需要的資訊，這個時候公眾號對于企業品牌露出是非常重要的通路和渠道。

正是由于這樣碎片化時間的特點，它使我們企業一個銷售的行為更加的場景化、更加連續化，相對于傳統的方式而言，它拓展了我們的渠道。通過公眾號里面各種插件和一系列的渠道起到了拉新的作用，基于這種公眾號屬性很有特點的一個模塊。企業把握移動互聯網的脈搏，使企業的營銷方案以最新的姿態出現在消費者或者用戶面前，保持這樣的一個互動，增加消費者的黏性。

現在說一下企業在這方面的困惑，剛才新京報的一個朋友在門口跟我說，上次參加完微信的一個會，微信表示它給大家已經搭了很多平臺，大家隨意用就好了，可是大家真的很迷茫，怎么用？為什么會有這樣的問題，是因為微信給了一個接口，如果企業真的運用到自己的企業，會進行開放，可是對于這樣的企業而言，大家并沒有相應的技術支持或者沒有這樣的技術基礎。

在我看來，大家企業微信公眾號存在三個問題，人、錢和時間。首先找專業的人、需要時間、需要錢來搭建你這個隊伍。人就不用說了，因為沒有人怎么做事兒呢，甚至很多企業往往把自己公司的網管、產品甚至前臺招來做這樣的事情，專業性就不用說了。另外需要時間，一個企業至少需要以月為單位的時間才能開發出一套可能行得通的東西，我說的只是可能，實際上大家都不是專業做這個事情的，做的最后東西呈現出來的效果未必能達到你最后心里的標準。最后是錢，需要時間和錢完成技術的搭建，花這樣的精力和時間是否值得做這樣的事情。剛才也說了，我們搭建工程的目的并不是為了搭建工程，而是為了提供這樣的窗口和外部世界發生聯系，和我們的客戶發生聯系，使老客戶、新客戶能夠不斷的到我們的企業這個平臺和我們做交互。

企業最重要的幾個問題是，資金有限、人力有限，以及如何使我的這些人到我的窗口來。正是因為大家有這樣的困惑，所以我們很快平臺實時的推出了快云的產品，它是一個免費優質的公眾號的管理平臺，為各個企業量身定制打造公眾號版的CRM系統，如果說微信給我們的企業提供的公眾號是一個毛坯房的話，快云這樣的產品給我們做的各個房間的格局布置裝修等一系列的事情。

快云提供了三個核心功能：基礎平臺功能升級、特色應用及插件以及解決方案。基礎平臺的功能升級不用多說，這樣的平臺都會提供的東西，我們對這個有很多增強，一個帳號無限制管理企業下面所有的公眾號的賬戶，能夠把現在市面上所有的圖文資源都整合到資源庫里面，不用再到另外的平臺編輯你們的內容，方便企業迅速形成圖文進行推廣。我們讓整個團隊為他們進行服務，能夠24小時實時的響應在整個公眾號搭建過程中所遇到的問題。我們有一個增強的數據統計平臺，進一步的挖掘潛在目標用戶的價值，使我們不斷的優化運營策略，推出相應的活動，進行一些拉新、拉活。

插件部分市面上有很多，基于微信的框架和外延大家都在開發這一塊，對于整個產品360度的展示，為網站也好、調查微表單也好、H5也好，都是提供給企業主做用戶的交互、拉新拉活等活動，比如說我們的紅包，實際上可以設置很多出發條件去滿足你所需要推出活動的方式，就像剛才張總說的一樣，怎么讓客戶購買更好的產品，這就是工具。

整個快云平臺最特色的就是現在提到給各個行業做的模板，酒店、醫療汽車、餐飲等各個行業提供專屬定制化的模板，按照月份不斷的更新和增加。關于酒店，今天上午，安美集團已經講了，在微網站做了很精辟的分享，我們酒店行業也是聯合了安美集團給他旗下全國2876家酒店搭建了一套完整公眾號的體驗方案，安美是國內一家很出名的酒店智能化服務的服務提供商，也是前段時間和我們達成戰略合作，通過我們的需求調研，給到他旗下像萬豪、洲際、喜達屋這樣的集團提供解決方案，360度全景展示，實時預訂和查詢等等一系列的功能。在這種基礎上，大大提升五星級酒店客戶的滿意度和交互度，并進一步產生很多數據的需求，從而推進整個服務的升級。

我們再來看下面一個案例，教育。我認為教育是和咱們互聯網聯系的比較早的一個行業，像泰課在線這樣一個國內最早做互聯網教育的企業，在線提供很多的課程，包括一些公開課、泰斗課等等，實際上為了增加客戶體驗，前段時間我們為他們搭建了專屬的公眾號平臺，咨詢、問答以及在線客戶打通了現在老師、課程和客戶之間的紐帶，為客戶提供更多場景化的一些應用和需求。

最后，我們看一下在紙媒行業的應用，這本雜志我相信很多人應該見過，《電腦愛好者》，它是很有年份、很有歷史的一份雜志，曾經是很強的意見領袖，現在也面臨很大的挑戰，正是基于此，想在移動互聯網做一個彎道超車，我們和他進行很深入地交流，并且對他的特點進行了全新的分析，為他搭建了商城、論壇一系列的定制開發。因為剛才提到很快是一個基于全網最大的流量分發平臺，我們還為他提供了流量的優勢，為他進一步拉新、拉活以及喚醒城市用戶提供了整體的解決方案。從現在運營情況來看，數值在不斷的網上漲。

快云的優勢，快云背后依靠的是整個生態，而并非功能和它的一些作用，雖然功能和插件各方面很強大，我認為整個快云的優勢是建立在我們流量分發的平臺優勢上，就能夠為我們的企業提供一站式的標準化服務，包括我們標準和定制化的模塊，專業的團隊和我們的生態資源，也是它很強有力的后盾。我們看一看快云資源優勢，剛才提到了，快云是給我們整個企業搭建公眾號的一個架構，通過我們流量分發的平臺，給我們的企業公眾號做一個導流，導流的同時使用戶質量和活躍度不斷提升，實現商業和流量變現的雙價值。我們為這樣的快云系統去植入大數據的插件，區別于傳統功能性為主的平臺，企業主通過這樣的數據插件實時了解自己在這個行業里面的情況，知道它的競品處于什么樣的狀態，并且知道用戶的行為，數據化的運用自身用戶的表征優化自己運營組合和策略。

最后，我們來看一看關于快云的套餐介紹。區別于其他以各種搭建為主要目的的平臺和產品，我們倒希望盡可能的釋放我們的創業公司或者公司最大的一個需求，不僅僅是公司在前期有很大的成本負擔，我們會對整個基礎的部分進行免費，對未來的插件模塊進行收費，因為企業在不同的發展階段，需求是不一致的，當發展一定程度的時候才需要特色化用戶拉新、拉活，針對企業不同的階段給他做一個更為好的方案，盡可能甩開手腳來做。包括用戶剛才提到的訂閱號、服務號，全場景的應用場景，使我們的用戶應用在所有的企業所承載的載體上面。

我們現在也在借助代理商的方式向全國推廣這樣的產品，包括技術支持、培訓，全網品牌的支持，使我們的代理商能夠實時的把我們產品優勢和產品功能能夠實時共享到創業企業中去。這是快云的合作，如果大家有需要可以拍照，我們事后聯系一下。

很多人會想為什么你們品牌叫很快？我們也是取自于一句俗語，天下武功為快不破，去年8月份成立，形成自己六大業務平臺大家可以想象成一個廣告聯盟，我們認為一個一個的公眾號是我們的微站系統，聚集起來使這個流量進行變現，快云就是我們今天介紹的公眾號搭建以及智能化管家的服務，幫大家搭建這樣的公眾號系統。快速主要為我們的快云以及快云提供客戶分析以及快云精準的廣告投放所服務，并且能夠提供這樣的榜單，為投資者了解這個行業的公眾號發展的情況以及投資。三個TOC平臺，快玩的游戲渠道以及看臉啦的社交平臺，下一步把直播的插件植入到各家公眾號，使未來真正進入全民直播的時代而不是一個全民直播平臺的時代，使我們的粉絲真正聚集到自己的公眾號，而不是聚集在別人的平臺上。謝謝各位聆聽！