一張圖進階 RocketMQ

前言

三此君看了好幾本書，看了很多遍原始碼整理的一張圖進階 RocketMQ 圖片，關於 RocketMQ 你只需要記住這張圖！覺得不錯的話，記得點贊關注哦。

【重要】視訊在 B 站同步更新，歡迎圍觀，輕輕鬆鬆漲姿勢。一張圖進階 RocketMQ-通訊機制（視訊版）
點選檢視【bilibili】

本文是「一張圖進階 RocketMQ」第 4 篇，對 RocketMQ 不瞭解的同學可以先看看前面三期：

上一期分享了 RocketMQ 生產者啟動流程及同步訊息傳送流程，我們知道了在通訊層是基於 Netty 將訊息傳遞給 Broker 進行儲存的。如果對 Netty 完全不瞭解我們就很難真正理解 RocketMQ，所以今天我們簡單的聊一聊 Netty 基本流程，然後分析 RocketMQ 的通訊機制，最後通過非同步訊息傳送來串聯 RocketMQ 通訊機制。

Netty 介紹

Netty 有很多概念，等介紹完概念大家都困了，我們就不過多介紹了，直接結合範例來看看 Netty 的基礎流程，能夠幫助我們更好的理解 RocketMQ 即可。

Netty 伺服器端啟動初始化兩個執行緒組 BossGroup & WorkerGroup，分別用於處理使用者端連線及網路讀寫。
Netty 使用者端啟動初始化一個執行緒組，用於處理請求及返回結果。
使用者端 connect 到 Netty 伺服器端，建立用於 傳輸資料的 Channel。
Netty 伺服器端的 BossGroup 處理使用者端的連線請求，然後把剩下的工作交給 WorkerGroup。
連線建立好了，使用者端就可以利用這個連線傳送資料給 Netty 伺服器端。
Netty WorkerGroup 中的執行緒使用 Pipeline(包含多個處理器 Handler) 對資料進行處理。
Netty 伺服器端的處理完請求後，返回結果也經過 Pipeline 處理。
Netty 伺服器端通過 Channel 將資料返回給使用者端。
使用者端通過 Channel 接收到資料，也經過 Pipeline 進行處理。

Netty 範例

我們先用 Netty 實現一個簡單的伺服器端/使用者端通訊範例，我們是這樣使用的，那 RocketMQ 基於 Netty 的通訊也應該是這樣使用的，不過是在這個基礎上封裝了一層。主要關注以下幾個點：伺服器端什麼時候初始化的，伺服器端實現的 Handler 做了什麼事？使用者端什麼時候初始化的，使用者端實現的 Handler 做了什麼事？
Netty 伺服器端初始化：初始化的程式碼很關鍵，我們要從原始碼上理解 RocketMQ 的通訊機制，那肯定會看到類似的程式碼。根據上面的流程來看，首先是範例化 bossGroup 和 workerGroup，然後初始化 Channel，從程式碼可以看出我們是在 Pipeline 中新增了自己實現的 Handler，這個 Handler 就是業務自己的邏輯了，那 RocketMQ 要處理資料應該也需要實現相應的 Handler。

public class MyServer {
    public static void main(String[] args) throws Exception {
        //建立兩個執行緒組 boosGroup、workerGroup
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //建立伺服器端的啟動物件，設定引數
            ServerBootstrap bootstrap = new ServerBootstrap();
            //設定兩個執行緒組boosGroup和workerGroup
            bootstrap.group(bossGroup, workerGroup)
                //設定伺服器端通道實現型別    
                .channel(NioServerSocketChannel.class)
                //使用匿名內部類的形式初始化Channel物件    
                .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            //給pipeline管道新增處理器
                            socketChannel.pipeline().addLast(new MyServerHandler());
                        }
                    });//給workerGroup的EventLoop對應的管道設定處理器
            //繫結埠號，啟動伺服器端
            ChannelFuture channelFuture = bootstrap.bind(6666).sync();
            //對關閉通道進行監聽
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

實現自定義的伺服器端處理器 Handler：自定義的 Handler 需要實現 Netty 定義的 HandlerAdapter，當有可讀事件時就會呼叫這裡的 channelRead() 方法。等下我們看 RocketMQ 通訊機制的時候留意RocketMQ 自定義了哪些 Handler，這些 Handler 有做了什麼事。

/**
 * 自定義的Handler需要繼承Netty規定好的 HandlerAdapter 才能被Netty框架所關聯，有點類似SpringMVC的介面卡模式
 **/
public class MyServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //獲取使用者端傳送過來的訊息
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("收到" + ctx.channel().remoteAddress() + "傳送的訊息：" + byteBuf.toString(CharsetUtil.UTF_8));
        //傳送訊息給使用者端
        ctx.writeAndFlush(Unpooled.copiedBuffer("伺服器端已收到訊息，記得關注三此君，記得三連", CharsetUtil.UTF_8));
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        //發生異常，關閉通道
        ctx.close();
    }
}

Netty 使用者端初始化：Netty 使用者端，在 RocketMQ 中對應了 Producer/Consumer。在 Producer 啟動中有一步是啟動通訊模組服務，其實就是初始化 Netty 使用者端。使用者端也需要先範例化一個 NioEventLoopGroup，然後將自定義的 handler 新增到 Pipeline，還有很重要的一步是我們需要 connect 連線到 Netty 伺服器端。

public class MyClient {
    public static void main(String[] args) throws Exception {
        NioEventLoopGroup eventExecutors = new NioEventLoopGroup();
        try {
            //建立bootstrap啟動引導物件，設定引數
            Bootstrap bootstrap = new Bootstrap();
            //設定執行緒組
            bootstrap.group(eventExecutors)
                //設定使用者端的Channel實現型別    
                .channel(NioSocketChannel.class)
                //使用匿名內部類初始化 Pipeline
                .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //新增使用者端Channel的處理器
                            ch.pipeline().addLast(new MyClientHandler());
                        }
                    })
            //connect連線伺服器端
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6666).sync();
            //對Channel關閉進行監聽
            channelFuture.channel().closeFuture().sync();
        } finally {
            //關閉執行緒組
            eventExecutors.shutdownGracefully();
        }
    }
}

實現自定義的使用者端處理器 Handler：使用者端處理器也繼承自 Netty 定義的 HandlerAdapter，當 Channel 變得可讀的時候（伺服器端資料返回）會呼叫我們自己實現的 channelRead()。

public class MyClientHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //傳送訊息到伺服器端
        ctx.writeAndFlush(Unpooled.copiedBuffer("三此君，我正在看 RocketMQ 生產者傳送訊息~", CharsetUtil.UTF_8));
    }
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //接收伺服器端傳送過來的訊息
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("收到三此君的訊息，我一定會三連的" + ctx.channel().remoteAddress() + byteBuf.toString(CharsetUtil.UTF_8));
    }
}

RocketMQ 通訊流程

RocketMQ 通訊模組基於 Netty 實現，總體程式碼量不多。主要是 NettyRemotingServer和NettyRemotingClient，分別對應通訊的伺服器端和使用者端。根據前面的 Netty 範例，我們要理解 RocketMQ 如何基於 Netty 通訊，只需要知道 4 個地方：NettyRemotingServer 如何初始化，NettyRemotingClient 初始化，如何基於 NettyRemotingClient 傳送訊息，無論是使用者端還是伺服器端收到資料後都需要 Handler 來處理。

Broker/NameServer 需要啟動 Netty 伺服器端。Broker 我們後面會進一步分析，只需要知道 Broker 啟動的時候會呼叫 NettyRemotingServer.start() 方法初始化 Netty 伺服器。主要做了 4 件事：設定 BossGroup/WorkerGroup NioEventLoopGroup 執行緒組，設定 Channel，新增 NettyServerHandler，呼叫 serverBootstrap.bind() 監聽埠等待使用者端連線。
Producer/Consumer 需要啟動 Netty 使用者端，在生產者啟動流程中 MQClientInstantce 啟動通訊服務模組，其實就是呼叫NettyRemotingClient.start() 初始化 Netty 使用者端。主要做了 3 件事：設定使用者端 NioEventLoopGroup 執行緒組，設定 Channel，新增 NettyClientHandler。
使用者端設定了 Channel，但是 Channel 還沒有建立，因為 Channel 肯定要和具體的 Server IP Addr 關聯。在同步訊息傳送流程中，呼叫 NettyRemoteClient.invokeSync()，從 channelTables 快取中獲取或者建立一個新的 Channel，其實就是呼叫 bootstrap.connect() 連線到 NettyServer，建立用於通訊的 Channel。
有了 Channel 後，Producer 呼叫 Channel.writeAndFlush() 將資料傳送給伺服器。NettyRemotingServer WorkerGroup 處理可讀事件，呼叫 NettyServerHandler 處理資料。
NettyServerHandler 呼叫 processMessageReceived方法。processMessageReceived 方法做了什麼呢？通過傳入的請求碼 RequestCode 區別不同的請求，不同的請求定義了不同的 Processor。例如，是生產者存入訊息使用 SendMessageProcessor，查詢訊息使用 QueryMessageProcessor，拉取訊息使用 PullMessageProcessor。這些 Processor 在伺服器端初始化的時候，以 RequestCode 為 Key 新增到 Processor 快取中。processMessageReceived 就是根據 RequeseCode 獲取不同的 Processor，處理完後把結果返回給 NettyRemotingClient。
NettyRemotingClient 收到可讀事件，呼叫 NettyClientHandler 處理返回結果。NettyClientHandler也呼叫processMessageReceived 處理返回結果。processMessageReceived 從以 opaque 為 key ResponseTables 快取衝取出 ResponseFuture，將返回結果設定到 ResponseFuture。同步訊息則執行 responseFuture.putResponse()，非同步呼叫執行回撥。

非同步傳送

除了同步訊息傳送，RocketMQ 還支援非同步傳送。我們只需要在前面是範例中稍作修改就會得到一個非同步傳送範例，最大的不同在於傳送的時候傳入 SendCallback 接收非同步返回結果回撥。

public class AsyncProducer {
    public static void main(String[] args) throws Exception {
        // 範例化訊息生產者Producer
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
        // 設定NameServer的地址
        producer.setNamesrvAddr("localhost:9876");
        // 啟動Producer範例
        producer.start();
        // 建立訊息，並指定Topic，Tag和訊息體
        Message msg = new Message("Topic1","Tag", "Key", "Hello world".getBytes("UTF-8")); 
        // SendCallback 接收非同步返回結果的回撥
        producer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("關注呀！！！%-10d OK %s %n", index,sendResult.getMsgId());
            }
            @Override
            public void onException(Throwable e) {
                System.out.printf("三連呀！！！%-10d Exception %s %n", index, e);
                e.printStackTrace();
            }
        });
        // 如果不再傳送訊息，關閉Producer範例。
        producer.shutdown();
    }
}

同步傳送個非同步傳送主要的過程都是一樣的，不同點在於同步訊息呼叫 Netty Channel.writeAndFlush 之後是 waitResponse 等待 Broker 返回，而非同步訊息是呼叫預先定義好的回撥函數。

非同步訊息和同步訊息整體差不多，可以說在基於 Netty 實現非同步訊息比同步訊息還要簡單一下，我們這裡主要來看一些不同點：

呼叫 DefaultMQProducer 非同步傳送介面需要我們定義 SendCallback 回撥函數，在執行成功或者執行失敗後回撥。
DefaultMQProducerImp 中的 send 方法會將非同步傳送請求封裝成 Runable 提交到執行緒池，然後業務執行緒就直接返回了。
sendDefaultImpl 計算重試同步和非同步訊息有區別，非同步訊息在這裡不會重試，而是在後面結果返回的時候通過遞迴重試。
跟著呼叫鏈到 sendMessageAsync 方法，需要注意的是這裡構建了 InvokeCallback 範例，ResponseFuture 會持有該範例，Netty 結果返回後呼叫該範例的方法。
下面就是正常的 Netty 資料傳送流程，直到 Broker 處理完請求，返回結果。NettyRemotingClient 處理可讀事件，NettyClientHandler 處理返回結果，呼叫 ResponseFuture.executeInokeCallback，進而呼叫 InvokeCallback.operationComplete.
如果 Broker 返回結果是成功的，則封裝返回結果 SendResult，並回撥業務實現的 SendCallback.onSucess 方法，更新容錯項。
如果 Broker 返回失敗，或出現任何異常則執行重試，重試超過 retryTimesWhenSendFailed 次則回撥業務定義的 SendCallback.onException方法。

總結

以上就是 RocketMQ 訊息傳送的主要內容，我們簡單的總結下：

Netty：BossGroup 處理使用者端連線請求，生成 ServerSocketChannel 註冊到 WorkerGroup，WorkerGroup 處理網路讀寫請求，呼叫 Channel 對應的 Pipeline 處理請求，Pipeline 中有很多 ChannelHandler 對請求進行處理。
通訊機制：基於 Netty 實現，只需要留意 NettyRemotingServer/NettyRemotingClient 的初始化，並且在通道變得可讀/可寫時，會呼叫 NettyServerHandler/NettyClienthandler 進行處理。
同步非同步：同步和非同步訊息大同小異，只是同步訊息通過 Netty 傳送請求後會執行 ResponseFuture.waitResponse() 阻塞等待，而非同步訊息傳送請求後不會等待，請求返回回撥用 SendCallback 相應的方法。

以上就是今天全部的內容，如果覺得本期的內容對你有用的話記得點贊、關注、轉發收藏，這將是對我最大的支援。如果你需要 RocketMQ 相關的所有資料，可以評論區留言，或者關注三此君的公眾號，回覆 mq 即可。
訊息已經傳送給了 Broker，下一期我們將來看看Broker 是如何儲存訊息的，RocketMQ 如何支援百萬級的吞吐量？感謝觀看，我們下期再見

參考文獻

RocketMQ 官方檔案
RocketMQ 原始碼
丁威, 周繼鋒. RocketMQ技術內幕：RocketMQ架構設計與實現原理. 機械工業出版社, 2019-01.
李偉. RocketMQ分散式訊息中介軟體：核心原理與最佳實踐. 電子工業出版社, 2020-08.
楊開元. RocketMQ實戰與原理解析. 機械工業出版社, 2018-06.

前 言

Netty 介紹

Netty 範例

RocketMQ 通訊流程

非同步傳送

總結

參考文獻

前言