RocketMQ如何保证一定发送成功？

RocketMQ提供了三种消息发送方式：同步发送、异步发送和单向发送。每种方式都有其特定的应用场景和特点，下面我们详细介绍它们的原理和使用方法。 ### 1. 同步发送 #### 原理 同步发送是指消息发送方发出一条消息后，会在收到服务端同步响应之后才发送下一条消息的通讯方式。  #### 应用场景 此种方式应用场景非常广泛，例如重要通知邮件、报名短信通知、营销短信系统等。 #### 示例代码 ```java import com.aliyun.openservices.ons.api.*; import java.util.Date; import java.util.Properties; public class ProducerTest { public static void main(String[] args) { Properties properties = new Properties(); properties.put(PropertyKeyConst.AccessKey, "XXX"); properties.put(PropertyKeyConst.SecretKey, "XXX"); properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000"); properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXX"); Producer producer = ONSFactory.createProducer(properties); producer.start(); for (int i = 0; i < 100; i++) { Message msg = new Message( "TopicTestMQ", "TagA", "Hello MQ".getBytes() ); msg.setKey("ORDERID_" + i); try { SendResult sendResult = producer.send(msg); if (sendResult != null) { System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId()); } } catch (Exception e) { System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic()); e.printStackTrace(); } } producer.shutdown(); } } ``` ### 2. 异步发送 #### 原理 异步发送是指发送方发出一条消息后，不等服务端返回响应，接着发送下一条消息的通讯方式。消息队列RocketMQ版的异步发送，需要实现异步发送回调接口（`SendCallback`）。  #### 应用场景 异步发送一般用于链路耗时较长，对响应时间较为敏感的业务场景，例如视频上传后通知启动转码服务，转码完成后通知推送转码结果等。 #### 示例代码 ```java import com.aliyun.openservices.ons.api.*; import java.util.Properties; import java.util.concurrent.TimeUnit; public class ProducerTest { public static void main(String[] args) throws InterruptedException { Properties properties = new Properties(); properties.put(PropertyKeyConst.AccessKey, "XXX"); properties.put(PropertyKeyConst.SecretKey, "XXX"); properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000"); properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXX"); Producer producer = ONSFactory.createProducer(properties); producer.start(); Message msg = new Message( "TopicTestMQ", "TagA", "Hello MQ".getBytes() ); msg.setKey("ORDERID_100"); producer.sendAsync(msg, new SendCallback() { @Override public void onSuccess(final SendResult sendResult) { System.out.println("send message success. topic=" + sendResult.getTopic() + ", msgId=" + sendResult.getMessageId()); } @Override public void onException(OnExceptionContext context) { System.out.println("send message failed. topic=" + context.getTopic() + ", msgId=" + context.getMessageId()); } }); TimeUnit.SECONDS.sleep(3); producer.shutdown(); } } ``` ### 3. 单向发送 #### 原理 单向发送是指发送方只负责发送消息，不等待服务端返回响应且没有回调函数触发。此方式发送消息的过程耗时非常短，一般在微秒级别。  #### 示例代码 ```java import com.aliyun.openservices.ons.api.*; import java.util.Properties; public class ProducerTest { public static void main(String[] args) { Properties properties = new Properties(); properties.put(PropertyKeyConst.AccessKey, "XXX"); properties.put(PropertyKeyConst.SecretKey, "XXX"); properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000"); properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXX"); Producer producer = ONSFactory.createProducer(properties); producer.start(); for (int i = 0; i < 100; i++) { Message msg = new Message( "TopicTestMQ", "TagA", "Hello MQ".getBytes() ); msg.setKey("ORDERID_" + i); producer.sendOneway(msg); } producer.shutdown(); } } ``` ### 三种发送方式的对比 | 发送方式 | 应用场景 | 优点 | 缺点 | | -------- | -------- | ---- | ---- | | 同步发送 | 重要通知邮件、报名短信通知、营销短信系统等 | 确保消息成功发送 | 耗时较长 | | 异步发送 | 链路耗时较长，对响应时间较为敏感的业务场景 | 响应时间快，适用于高并发场景 | 需要实现回调接口，处理复杂 | | 单向发送 | 只关心发送而不关心响应的场景 | 发送过程耗时短，性能高 | 不能确认消息是否成功送达 | 通过以上三种方式的介绍及代码示例，可以根据具体业务需求选择合适的消息发送方式，以确保消息发送成功并满足系统的性能需求。

### RocketMQ如何保证消息不丢失 RocketMQ在分布式系统中，通过各种策略和机制，最大限度地保证消息在生产、存储和消费阶段的可靠性。下面详细介绍RocketMQ在这三个阶段的具体实现。 ### 一、发送端消息可靠性 消息发送一般有以下几种方式：同步发送、异步发送以及单向发送。根据具体业务需求选择不同的发送方式以保证消息的可靠性。 #### 1. 同步发送 同步发送是指发送端在发送消息时，阻塞线程进行等待，直到服务器返回发送结果。发送端通过检查Broker返回的状态来判断消息是否持久化成功。发送超时或失败时，RocketMQ会默认重试2次。虽然这种方式不能保证消息只传输一次，但可以保证至少传输成功一次。 #### 2. 异步发送 异步发送是指发送端在发送消息时，传入回调接口实现类，调用发送接口后不会阻塞，发送方法会立即返回。通过回调函数接收发送结果，并根据结果信息判断是否需要重试以保证消息的可靠性。 #### 3. 单向发送 单向发送是指发送端发送完成后，调用发送接口立即返回，不返回发送结果。这种方式发送消息的过程耗时非常短，但可靠性较低，不推荐用于需要高可靠性的业务场景。 #### 4. 发送重试策略 RocketMQ架构中有多个Broker为某个Topic提供服务，Broker将其提供存储服务的Topic的元数据信息上报到NameServer。发送端在发送消息时，首先缓存Topic的路由信息并包含消息队列，如果没有缓存则向NameServer查询。查询到路由信息后，采用轮询策略选择相应的队列发送消息，发送失败则根据策略重试到其他Broker，从而提高消息发送的可靠性。 ### 二、存储端消息可靠性 #### 1. 存储结构 RocketMQ的消息存储结构如下： - 消息存储的最小单位是消息（Message）。 - 同一个Topic下的消息映射成多个逻辑队列。 - 不同Topic的消息按照到达Broker的先后顺序以Append方式添加至CommitLog。 #### 2. 存储可靠性挑战 存储端在存储消息时面临以下挑战： - Broker正常关闭 - Broker异常Crash - OS Crash - 机器掉电后恢复供电 - 机器无法开机（硬件损坏） - 磁盘设备损坏 RocketMQ通过单机刷盘机制和主从数据复制来保证存储可靠性。 #### 3. 刷盘策略 - **同步刷盘**：消息写入PageCache后立刻通知刷盘线程刷盘，等待刷盘完成后返回写成功状态。保证数据绝对安全，但吞吐量较低。 - **异步刷盘**：消息写入PageCache后立即返回写操作成功，PageCache中的消息积累到一定量或定时触发写入磁盘。吞吐量大，性能高，但PageCache中的数据可能丢失。 #### 4. 过期文件删除 为了避免内存和磁盘浪费，RocketMQ引入了文件过期删除机制，保持磁盘水位在一定水平，保证新写入消息的可靠存储。 ### 三、消费端消息可靠性 RocketMQ默认提供至少消费一次的语义，保证消息的可靠消费。 #### 1. 消费重试 消费者从RocketMQ拉取到消息后，返回CONSUME_SUCCESS表示消费完成。如果返回CONSUME_LATER，则按照不同的delayLevel进行重试，最长时间为2小时。消费满16次仍未成功的消息将进入死信队列。 #### 2. 死信队列 未成功消费的消息会被发送到死信队列，确保消息存储的可靠性。可以通过RocketMQ提供的接口从死信队列获取消息。 #### 3. 消息回溯 回溯消费指Consumer重新消费已经成功消费的消息。Broker存储端在向Consumer投递成功消息后仍需保留消息，按时间维度回退消费进度，保证消息可以重新消费。 ### 四、总结 RocketMQ通过以下机制保证消息的可靠性： - **发送端**：通过不同的发送方式和重试策略，保证消息至少传输成功一次。 - **存储端**：通过同步和异步刷盘策略以及主从数据复制，保证消息的可靠存储。 - **消费端**：通过至少消费一次的语义、消费重试和死信队列，保证消息的可靠消费。 通过全链路闭环的可靠性保证，RocketMQ最大限度地避免了消息丢失。