RocketMQ 如何保證發送消息不丟失？

作者：猿java 2024-08-06 09:55:25

本文分析了 RocketMQ 同步發送、異步發送和單向發送三種方式的原理、優缺點以及使用場景，并且分析了每種方式涉及到的核心源碼。

在 RocketMQ 中，有 3種簡單的消息發送方式：同步發送、異步發送和單向發送。這篇文章，我們將詳細分析這三種發送方式的原理、優缺點、使用場景以及使用該方式是否會丟失數據。

本文源碼基于: Apache RocketMQ release-5.2.0

一、同步發送

1.原理分析

在同步發送模式下，RocketMQ 默認采用同步刷盤方式，當生產者將消息發送到 Broker 后，會等待 Broker 的響應（默認超時 5分鐘），Broker 接收消息后，會將其寫入內存緩存，并進行刷盤操作。因此，如果 Broker 響應成功，代表消息一定成功寫入磁盤。

同步發送主要涉及以下幾個步驟：

創建Producer：創建一個Producer對象；
創建消息：創建一個Message對象，設置Topic、Tag標簽和消息體；
發送消息：調用DefaultMQProducer的send方法；
等待響應：發送方會阻塞等待服務器的響應，直到收到確認消息；

如下示例代碼為一個完整的同步發送流程：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;

public class SyncProducerTest {
  public static void main(String[] args) throws Exception {
    // 1、創建 producer，設置組名為 SyncGroupTest
    DefaultMQProducer producer = new DefaultMQProducer("SyncGroup");
    // 2、指定 NameServer的地址，以獲取 Broker路由地址
    producer.setNamesrvAddr("x.x.x.x:9876");
    // 3、啟動 producer
    producer.start();
    // 4、創建消息，并指定 Topic，Tag和消息體
    Message msg = new Message("SyncTopic", "sync", "SyncMessage".getBytes("UTF-8"));
    // 5、發送同步消息
    SendResult sendResult = producer.send(msg);
    // 6、通過 sendResult 判斷消息是否成功送達
    System.out.printf("message send result:" + sendResult);
    // 7、關閉 Producer
    producer.shutdown();
  }
}

RocketMQ 的同步發送主要涉及以下幾個關鍵源碼類和方法：

DefaultMQProducer：生產者類，負責發送消息。
MQClientAPIImpl#sendMessage：底層消息發送實現。
NettyRemotingClient#invokeSync：通過 Netty 實現網絡通信。
Broker 端的 SendMessageProcessor：處理發送請求。

源碼參考：org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#send(Message msg)

2.優缺點

優點：

簡單易用。
可靠性高，發送方可以確認消息是否成功發送，一旦發送成功，消息就已經寫入磁盤，消息不會丟失。

缺點：

延遲較高，需要等待服務器的響應。
吞吐量可能受限于網絡延遲和服務器性能。

3.使用場景

適用于對消息可靠性要求較高的場景，如訂單系統、金融交易、重要的消息通知等。

二、異步發送

1.原理分析

在異步發送模式下，RocketMQ 默認采用異步刷盤方式，當生產者發送消息到 Broker 后，消息寫入內存緩存成功后，Broker 立即返回響應（默認超時 5分鐘），后臺線程再異步將消息批量寫入磁盤。因此，這種方式提高了系統的吞吐量和性能，但在系統崩潰時可能會丟失部分未刷盤的消息。

異步發送主要涉及以下幾個步驟：

創建Producer：創建一個Producer對象；
創建消息：同樣創建一個Message對象。
發送消息：調用DefaultMQProducer的send方法，傳遞一個SendCallback回調對象。
處理響應：回調函數會在消息發送成功或失敗時被調用。

如下示例代碼為一個完整的異步發送流程：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;

public class AsyncProducerTest {
  public static void main(String[] args) throws Exception {
    // 1、創建 producer，設置組名為 AsyncGroupTest
    DefaultMQProducer producer = new DefaultMQProducer("AsyncGroup");
    // 2、指定 NameServer的地址，以獲取 Broker路由地址
    producer.setNamesrvAddr("x.x.x.x:9876");
    // 3、啟動 producer
    producer.start();
    // 4、創建消息，并指定Topic，Tag和消息體
    Message msg = new Message("AsyncTopic","async", "AsyncMessage".getBytes("UTF-8"));
    // 5、發送異步消息，SendCallback是處理異步回調的方法
    producer.send(msg, new SendCallback() {
      @Override
      public void onSuccess(SendResult sendResult) {  // 成功回調
        System.out.println("message send success: " + sendResult);
      }
      @Override
      public void onException(Throwable throwable) {  // 失敗回調
        System.out.println("message send fail: " + throwable);
      }
    });
    // 6、關閉 Producer
    producer.shutdown();
  }
}

RocketMQ 的異步發送主要涉及以下幾個關鍵源碼類和方法：

DefaultMQProducer：生產者類，負責發送消息。
MQClientAPIImpl#sendMessage：底層消息發送實現。
NettyRemotingClient#invokeAsync：通過 Netty 實現網絡通信。
Broker 端的 SendMessageProcessor：處理發送請求。

源碼參考：org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#send(Message msg, SendCallback sendCallback)

2.優缺點

優點：

非阻塞，發送方可以繼續執行其他任務，提高吞吐量。
延遲較低，適用于對響應時間敏感的場景。

缺點：

實現復雜度較高，需要處理異步回調。
可靠性相對降低，需要處理失敗重試等問題。
無法保證發送出去的數據不丟失。

3.使用場景

適用于對響應時間要求較高的場景，如實時數據處理、日志采集、消費信息的推送等。

三、單向發送

1.原理分析

單向（OneWay）發送是一種只負責發送消息而不等待任何響應的方式。生產者將消息發送到 Broker 后（默認超時 5分鐘），不關心消息是否成功到達或被持久化，主要依賴 Broker 進行刷盤操作，單向發送通常與異步刷盤結合使用，以提高發送效率。

單向發送主要涉及以下幾個步驟：

創建Producer：創建一個Producer對象；
創建消息：創建一個Message對象。
發送消息：調用DefaultMQProducer的sendOneway方法。

如下示例代碼為一個完整的單向發送流程：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;

public class OneWayProducerTest {
  public static void main(String[] args) throws Exception {
    // 1、創建 producer，設置組名為 OneWayGroupTest
    DefaultMQProducer producer = new DefaultMQProducer("OneWayGroup");
    // 2、指定 NameServer的地址，以獲取 Broker路由地址
    producer.setNamesrvAddr("x.x.x.x:9876");
    // 3、啟動 producer
    producer.start();
    // 4、創建消息，并指定Topic，Tag和消息體
    Message msg = new Message("OneWayTopic","oneway", "OneWayMessage".getBytes("UTF-8"));
    // 5、發送單向消息
    producer.sendOneway(msg);
    // 6、關閉 Producer
    producer.shutdown();
  }
}

RocketMQ 的單向發送主要涉及以下幾個關鍵類和方法：

DefaultMQProducer：生產者類，負責發送消息。
MQClientAPIImpl#sendMessage：底層消息發送實現。
NettyRemotingClient#invokeOneway：通過 Netty 實現網絡通信。
Broker 端的 SendMessageProcessor：處理發送請求。

源碼參考：org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#sendOneway(Message msg)

2.優缺點

優點：

非常高效，延遲最低。
適用于對可靠性要求不高的場景。

缺點：

無法確認消息是否成功發送。
可靠性最低，消息可能丟失。

3.使用場景

適用于對可靠性要求不高的場景，如日志收集、監控數據上報等。

三種方式對比

發送方式	優點	缺點	使用場景
同步發送	可靠性高，簡單易用	延遲較高，吞吐量受限	訂單系統、金融交易、重要的消息通知等
異步發送	非阻塞，延遲較低	實現復雜度高，可靠性相對降低	實時數據處理、日志采集、消費信息的推送等
單向發送	高效，延遲最低	無法確認消息是否成功發送，可靠性最低	日志收集、監控數據上報等

如何選擇？

同步發送：消息發送后會等待服務器的響應，整個過程業務是阻塞等待的，適用于對可靠性要求高的場景，比如訂單系統、金融交易等。
異步發送：消息發送后，不等待服務器響應，而是通過回調函數處理響應，適用于對響應時間要求高的場景，比如實時數據處理、日志采集、消費信息的推送等
單向發送：：單向發送只負責發送消息而不等待任何響應的方式，也不需要對發送的狀態、結果負責，適用于對可靠性要求不高的場景，比如日志收集、監控數據上報等。

每種發送方式都有其適用的場景和優缺點，具體如何選擇，一定需要根據業務需求進行權衡。

總結

本文分析了 RocketMQ 同步發送、異步發送和單向發送三種方式的原理、優缺點以及使用場景，并且分析了每種方式涉及到的核心源碼。

通過上文的介紹可以知道同步發送方式可以保證消息發送時不丟，但是性能相對其他兩種方式差一些。

RocketMQ 是一款優秀的開源消息中間件，作為 Java程序員，建議多去閱讀它的源碼，吸收其中比較好的代碼思維。

責任編輯：趙寧寧來源：猿java

RocketMQ 消息發送