本文源码基于: 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程序员，建议多去阅读它的源码，吸收其中比较好的代码思维。
