在上一篇文章中，我們學習了RocketMQ的原理，以及RocketMQ中命名服務 ServiceName 的運作流程，本篇從消息的生産、消費來了解一條消息的生命周期。

1 消息生産

在RocketMQ中，消息生産指的是消息生産者往消息隊列中寫入資料的過程。因為業務場景的複雜性，RocketMQ架構設計了多種不同的寫入政策。下面先讨論幾種常見的場景：

同步發送：整個過程業務是阻塞等待的，消息發送之後等待 Broker 響應，得到響應結果之後再傳遞給業務線程。

-異步發送：調用RocketMQ 的 Async API，消息生産者隻要把消息發送任務放進線程池就傳回給業務線程。所有的邏輯處理、IO操作、網絡請求都由線程池處理，處理完成之後，調用業務程式定義好的回調函數來告知業務最終的結果。
OneWay（單向）發送：隻負責觸發對消息的發送，發送出即完成任務，不需要對發送的狀态、結果負責。
延遲發送：指定延遲的時間，在延遲時間到達之後再進行消息的發送。
批量發送：對于同類型、同特征的消息，可以聚合進行批量發送，減少MQ的連接配接發送次數，能夠顯著提升性能。

以下是生産者執行個體化啟動的過程：

1.1 消息發送步驟

一般情況下，我們發送消息，會使用預設的DefaultMQProducer類，經過以下幾個步驟實作：

建立消息生産者Producer，并設定Producer的GroupName（生産組）。
設定InstanceName（執行個體名稱），當你的業務需要啟用多個Producer的時候，使用不同的InstanceName來區分。
設定NameServer位址，這樣Producer才能從NameServer中得到路由資訊
完成其他的初始化配置，比如配置異常重試次數（降低消息丢失的可能性），通信子產品初始化等。
組裝消息對象，指定主題Topic、Tag和消息體Message 等資訊。
通過NameServer擷取到的Broker路由位址，将消息發送。

1.2 消息發生傳回狀态

消息發送之後，會相應的拿到回執。傳回對象中的狀态（SendResult.SendStatus）有4種，如下：

FLUSH_DISK_TIMEOUT 刷盤逾時

如果将Broker的刷盤政策設定成SYNC_FLUSH，那麼沒有在規定的時間完成刷盤則會報該錯誤。
FLUSH_SLAVE_TIMEOUT 主從同步逾時

主從模式下（也可以叫主備），Broker配置為SYNC_MASTER模式，如果沒有在設定時間内完成主從同步，則會報該錯誤。
SLAVE_NOT_AVAILABLE 未找到Slave Broker

主從模式下，且Broker配置為SYNC_MASTER，如果未找到Slave的Broker，則會報該錯誤。
SEND_OK

表示發送成功。

1.3 發送同步消息

實時同步消息是一種對可靠性、實時性要求比較高的場景，使用的也比較廣泛，比如：

重要的消息通知，比如驗證碼，不能超過太長時間推送，那樣可能失效
消費記錄确認
資料實時處理和推送等等

public class SyncProducerApplication {
    public static void main(String[] args) throws Exception {
        // 1、建立生産者producer，并指定生産者組名為 testSyncGroup
        DefaultMQProducer producer = new DefaultMQProducer("testSyncGroup");
        // 2、指定NameServer的位址，以擷取Broker路由位址
        producer.setNamesrvAddr("192.168.139.1:9876");
        // 3、啟動producer
        producer.start();
        // 4、建立消息，并指定Topic，Tag和消息體
        Message msg = new Message("testTopic","sync", "測試同步消息".getBytes("UTF-8"));
        // 5、發送消息到一個Broker
        SendResult sendResult = producer.send(msg);
        // 6、通過sendResult傳回消息是否成功送達
        System.out.printf("%s%n", sendResult);
        // 7、如果不再發送消息，關閉生産者Producer
        producer.shutdown();
    }
}

1.4 發送異步消息

我們知道，異步主要用于那些對實時響應不敏感的業務，可以容忍一定時間的等待，隻要能達到最終一緻性即可。

有時候為了在流量高峰期進行削峰和分流，緩解壓力，我們經常采用異步消息的發送模式。這種業務場景也很常見，比如：

消費資訊的推送，可能在你買單之後的幾分鐘才送達
資料統計、檔案打包下載下傳等需要長耗時的任務

public class AsyncProducerApplication {
    public static void main(String[] args) throws Exception {
        // 1、建立生産者producer，并指定生産者組名為 testAsyncGroup
        DefaultMQProducer producer = new DefaultMQProducer("testAsyncGroup");
        // 2、指定NameServer的位址，以擷取Broker路由位址
        producer.setNamesrvAddr("192.168.139.1:9876");
        // 3、啟動producer
        producer.start();
        // 4、建立消息，并指定Topic，Tag和消息體
        Message msg = new Message("testTopic","async", "測試異步消息".getBytes("UTF-8"));
        // 5、發送消息到一個Broker
        SendResult sendResult = producer.send(msg);
        // 6. 發送異步消息，SendCallback是處理異步回調的方法
        producer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {  // 成功回調
                System.out.println("success: " + sendResult);
            }
            @Override
            public void onException(Throwable throwable) {  // 失敗回調
                System.out.println("fail: " + throwable);
            }
        });
        // 7、如果不再發送消息，關閉生産者Producer
        producer.shutdown();
    }
}

1.5 單向發送消息

OneWay的模式主要用在Care發送結果的場景，隻要消息發送出去即完成任務，不需要對發送的狀态、結果負責。常見的使用場景如

普通日志記錄
非核心的埋點上報等

public class OneWayProducerApplication {
    public static void main(String[] args) throws Exception {
        // 1、建立生産者producer，并指定生産者組名為 testOneWayGroup
        DefaultMQProducer producer = new DefaultMQProducer("testOneWayGroup");
        // 2、指定NameServer的位址，以擷取Broker路由位址
        producer.setNamesrvAddr("192.168.139.1:9876");
        // 3、啟動producer
        producer.start();
        // 4、建立消息，并指定Topic，Tag和消息體
        Message msg = new Message("testTopic","oneway", "測試單向發送消息".getBytes("UTF-8"));
        // 5、發送消息到一個Broker
        producer.sendOneway(msg);
        // 6、如果不再發送消息，關閉生産者Producer
        producer.shutdown();
    }
}

1.6 發送延時消息

指定延遲的時間，在延遲時間到達之後再進行消息的發送。這種的使用場景也很多：

比如火車票訂購，送出了一個訂單就把車票給占位了，這時候可以發送一個延時确認的消息，15m 未付款，就要把該車票釋放，讓其他人去購買。
還比如購買了電影票，可以發送一個核銷資訊，在電影開場前15分鐘就無法退票了。

1.6.1 延時時間的使用限制

延時時間并不是随意指定的，Rocket源碼中指定了18種等級，分别代表不同的時間時長，如下：

// org/apache/rocketmq/store/config/MessageStoreConfig.java 的第198行
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

RocketMq不支援任意時間延時，需設定固定的延時等級，從1s到2h分别對應着等級1到18
可以使用setDelayTimeLevel(int level) 方法設定延時等級，level 從 0 開始

1.6.2 發送延時消息具體實作

通過下面的代碼，可以得到的結果是消費的時間點比資訊記錄的時間點延遲了1分鐘，這是因為我們在send的時候做了delay。

public class DelayProducerApplication {
    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException , UnsupportedEncodingException {
        // 1、建立生産者producer，并指定生産者組名為 testDelayGroup
        DefaultMQProducer producer = new DefaultMQProducer("testDelayGroup");
        // 2、指定NameServer的位址，以擷取Broker路由位址
        producer.setNamesrvAddr("192.168.139.1:9876");
        // 3、啟動producer
        producer.start();
        // 4、建立消息，并指定Topic，Tag和消息體
        Message msg = new Message("testTopic","delay", "測試延遲發送消息".getBytes("UTF-8"));
        // 5、設定延時等級4,對應1m,是以這個消息在一分鐘之後發送
        msg.setDelayTimeLevel(4);
        // 6、發送消息到一個Broker
        SendResult sendResult = producer.send(msg);
        // 7、通過sendResult傳回消息是否成功送達
        System.out.printf("%s%n", sendResult);
        // 8、如果不再發送消息，關閉生産者Producer
        producer.shutdown();
    }
}

1.7 發送批量消息

對于同類型、同特征的消息，可以聚合進行批量發送，減少MQ的連接配接發送次數，能夠顯著提升性能。
批量發送消息須有相同的topic，相同的waitStoreMsgOK，且不能是延時消息。

waitStoreMsgOK: 消息發送時是否等消息存儲完成後再傳回。

一批次的消息總大小不應超過4MB。

public class BatchProducerApplication {
    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, UnsupportedEncodingException {
        // 1、建立生産者producer，并指定生産者組名為 testBatchGroup
        DefaultMQProducer producer = new DefaultMQProducer("testBatchGroup");
        // 2、指定NameServer的位址，以擷取Broker路由位址
        producer.setNamesrvAddr("192.168.139.1:9876");
        // 3、啟動producer
        producer.start();
        // 4、建立消息清單，并指定Topic，Tag和消息體
        List<Message> messages = new ArrayList<>();
        String topic = "testTopic";
        messages.add(new Message(topic, "batch", "測試批量發送消息 0".getBytes("UTF-8")));
        messages.add(new Message(topic, "batch", "測試批量發送消息 1".getBytes("UTF-8")));
        messages.add(new Message(topic, "batch", "測試批量發送消息 2".getBytes("UTF-8")));

        // 5、發送消息到一個Broker
        SendResult sendResult = producer.send(messages);
        // 6、通過sendResult傳回消息是否成功送達
        System.out.printf("%s%n", sendResult);
        // 7、如果不再發送消息，關閉生産者Producer
        producer.shutdown();
    }
}

1.8 如何提升消息生産的性能

消息的發送一般是經過 client發送、Broker伺服器接收并處理、Broker伺服器傳回應答三個步驟。

如果我們想要提高消息生産的效率，一般有如下方法：

Oneway方式發送

Oneway方式發送用在一些性能要求高，可靠性要求低的場景下，比如日志采集，非核心的埋點上報等。Oneway方式發送請求無需應答，即将資料寫入用戶端的Socket緩沖區就傳回，不等待結果的傳回。

是以這種模式是極快的，可以把發送消息時長縮短至微秒級。
增加Producer的并發量，使用多個Producer同時發送

RocketMQ引入了一個并發視窗，在視窗内消息可以并發地寫入DirectMem中，然後異步地将連續資料寫入檔案系統。

順序執行CommitLog讓RocketMQ可以保持較高的寫入性能。
恰當的批量發送

對于同類型、同特征的消息，可以聚合進行批量發送，減少MQ的連接配接發送次數，能夠顯著提升性能。批量發送消息須有相同的topic，相同的waitStoreMsgOK，且不能是延時消息。

對于消息體的大小也要注意不能超過4MB。

根據阿裡内部調優後的性能測試報告，消息的寫入性能達到90萬+的TPS，我們可以朝着這個名額進行優化。

2 總結

本篇介紹了RocketMQ 消息生産與發送的幾種模式：

同步發送：整個過程業務是阻塞等待的，消息發送之後等待 Broker 響應，得到響應結果之後再傳遞給業務線程。
異步發送：調用RocketMQ 的 Async API，消息生産者隻要把消息發送任務放進線程池就傳回給業務線程。所有的邏輯處理、IO操作、網絡請求都由線程池處理，處理完成之後，調用業務程式定義好的回調函數來告知業務最終的結果。
OneWay（單向）發送：隻負責觸發對消息的發送，發送出即完成任務，不需要對發送的狀态、結果負責。
延遲發送：指定延遲的時間，在延遲時間到達之後再進行消息的發送。
批量發送：對于同類型、同特征的消息，可以聚合進行批量發送，減少MQ的連接配接發送次數，能夠顯著提升性能。

可以根據實際的業務場景選擇适當的發送模式。

為幫助開發者們提升面試技能、有機會入職BATJ等大廠公司，特别制作了這個專輯——這一次整體放出。

大緻内容包括了： Java 集合、JVM、多線程、并發程式設計、設計模式、Spring全家桶、Java、MyBatis、ZooKeeper、Dubbo、Elasticsearch、Memcached、MongoDB、Redis、MySQL、RabbitMQ、Kafka、Linux、Netty、Tomcat等大廠面試題等、等技術棧！

歡迎大家關注公衆号【Java爛豬皮】，回複【666】，擷取以上最新Java後端架構VIP學習資料以及視訊學習教程，然後一起學習，一文在手，面試我有。

每一個專欄都是大家非常關心，和非常有價值的話題，如果我的文章對你有所幫助，還請幫忙點贊、好評、轉發一下，你的支援會激勵我輸出更高品質的文章，非常感謝！

深入了解MQ：RocketMQ消息的發送模式

1 消息生産

1.1 消息發送步驟

1.2 消息發生傳回狀态

1.3 發送同步消息

1.4 發送異步消息

1.5 單向發送消息

1.6 發送延時消息

1.6.1 延時時間的使用限制

1.6.2 發送延時消息具體實作

1.7 發送批量消息

1.8 如何提升消息生産的性能

2 總結

繼續閱讀

RocketMQ生産者消費者部分源碼分析總結

Rocketmq 消費者組順序、平均消費實作原理

RocketMQ 5.x延時消息源碼分析(不包含時間輪)

rocketMq立即消費與延時消費

《深入了解Java虛拟機》讀書筆記：垃圾收集算法

RocketMQ高階使用（順序消息生産者保證政策消費者投遞政策消費者重試政策）RocketMQ高階使用

【部署】【二】制作 RocketMQ Console 控制台 Docker 鏡像環境準備和最終目标下載下傳源碼并編譯（本地）準備檔案（ Linux 主機）生成 Docker 鏡像啟動容器測試通過浏覽器檢視控制台

《深入了解Java虛拟機》讀書筆記：垃圾收集器

能源和環境問題是目前全球面臨的重要挑戰之一，新型Co-X電極材料能提供幫助？前言能源和環境問題是目前全球面臨的重要挑戰之

讀懂一本書至關重要。首先，需要通過百度百科等途徑全面了解書的内容。有些孩子可能不願意讀名著，因為看不懂。但是，通過百度百

深入了解AQS獨占鎖之ReentrantLock源碼分析

RocketMQ學習-服務安裝、及服務端指令學習

精通Spring 5.0：深入了解自動化配置的條件化注解

深入了解 HDFS（二）：Replica

rocketMQ安裝和使用文檔

聊聊 RocketMQ 名字服務