你好，我是猿java。

在 RocketMQ 中，有 3种简单的消息发送方式：同步发送、异步发送和单向发送。这篇文章，我们将详细分析这三种发送方式的原理、优缺点、使用场景以及使用该方式是否会丢失数据。

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

同步发送 {#同步发送}

原理分析 {#原理分析}

在同步发送模式下，RocketMQ 默认采用同步刷盘方式，当生产者将消息发送到 Broker 后，会等待 Broker 的响应（默认超时 5分钟），Broker 接收消息后，会将其写入内存缓存，并进行刷盘操作。因此，如果 Broker 响应成功，代表消息一定成功写入磁盘。

同步发送主要涉及以下几个步骤：

创建Producer ：创建一个Producer对象；
创建消息 ：创建一个Message对象，设置Topic、Tag标签和消息体；
发送消息 ：调用DefaultMQProducer的send方法；
等待响应：发送方会阻塞等待服务器的响应，直到收到确认消息；

如下示例代码为一个完整的同步发送流程：

|------------------------------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 | 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)

优缺点 {#优缺点}

优点：

简单易用。
可靠性高，发送方可以确认消息是否成功发送，一旦发送成功，消息就已经写入磁盘，消息不会丢失。

缺点：

延迟较高，需要等待服务器的响应。
吞吐量可能受限于网络延迟和服务器性能。

使用场景 {#使用场景}

适用于对消息可靠性要求较高的场景，如订单系统、金融交易、重要的消息通知等。

异步发送 {#异步发送}

原理分析 {#原理分析-1}

在异步发送模式下，RocketMQ 默认采用异步刷盘方式，当生产者发送消息到 Broker 后，消息写入内存缓存成功后，Broker 立即返回响应（默认超时 5分钟），后台线程再异步将消息批量写入磁盘。因此，这种方式提高了系统的吞吐量和性能，但在系统崩溃时可能会丢失部分未刷盘的消息。

异步发送主要涉及以下几个步骤：

创建Producer ：创建一个Producer对象；
创建消息 ：同样创建一个Message对象。
发送消息 ：调用DefaultMQProducer的send方法，但传递一个SendCallback回调对象。
处理响应：回调函数会在消息发送成功或失败时被调用。

如下示例代码为一个完整的异步发送流程：

|---------------------------------------------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 | 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)

优缺点 {#优缺点-1}

优点：

非阻塞，发送方可以继续执行其他任务，提高吞吐量。
延迟较低，适用于对响应时间敏感的场景。

缺点：

实现复杂度较高，需要处理异步回调。
可靠性相对降低，需要处理失败重试等问题。
无法保证发送出去的数据不丢失。

使用场景 {#使用场景-1}

适用于对响应时间要求较高的场景，如实时数据处理、日志采集、消费信息的推送等。

单向发送 {#单向发送}

原理分析 {#原理分析-2}

单向（OneWay）发送是一种只负责发送消息而不等待任何响应的方式。生产者将消息发送到 Broker 后（默认超时 5分钟），不关心消息是否成功到达或被持久化，主要依赖 Broker 进行刷盘操作，单向发送通常与异步刷盘结合使用，以提高发送效率。

单向发送主要涉及以下几个步骤：

创建Producer ：创建一个Producer对象；
创建消息 ：创建一个Message对象。
发送消息 ：调用DefaultMQProducer的sendOneway方法。

如下示例代码为一个完整的单向发送流程：

|------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 | 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}

优点：

非常高效，延迟最低。
适用于对可靠性要求不高的场景。

缺点：

无法确认消息是否成功发送。
可靠性最低，消息可能丢失。

使用场景 {#使用场景-2}

适用于对可靠性要求不高的场景，如日志收集、监控数据上报等。

3种方式对比 {#3种方式对比}

| 发送方式 | 优点 | 缺点 | 使用场景 | |------|-----------|--------------------|----------------------| | 同步发送 | 可靠性高，简单易用 | 延迟较高，吞吐量受限 | 订单系统、金融交易、重要的消息通知等 | | 异步发送 | 非阻塞，延迟较低 | 实现复杂度高，可靠性相对降低 | 实时数据处理、日志采集、消费信息的推送等 | | 单向发送 | 高效，延迟最低 | 无法确认消息是否成功发送，可靠性最低 | 日志收集、监控数据上报等 |