RocketMQ 如何保证消息不丢失？

server/2025/1/12 17:44:19/

RocketMQ 的消息想要确保不丢失，需要生产者、消费者以及 Broker 的共同努力，缺一不可。

生产者（Producer）

1、发送方式：选择同步发送

同步发送：发送消息后，需要阻塞等待 Broker 确认收到消息，生产者才能拿到返回的 SendResult
异步发送：Producer 首先构建一个向 broker 发送消息的任务，把该任务提交给线程池，等执行完该任务时，回调用户自定义的回调函数，执行处理结果。

2、重试机制

生产者因为网络故障、服务异常等原因导致调用， RocketMQ 内置请求重试逻辑，默认重试 3 次。
重试 3 次意思是一共会发 4 次消息，1 次原始消息，3 次重试消息。

Broker

1、刷盘策略：选择同步刷盘

同步刷盘：如果想要保证消息不丢失，可以将消息保存机制修改为同步刷盘，这样，Broker会在同步请求中把数据保存在磁盘上，确保保存成功后再返回确认结果给生产者。
异步刷盘：异步发送的话就需要生产者重写 SendCalback 的 onSuccess 和 onException 方法，用于给 Broker 进行回调。在方法中实现消息的确认或者重新发送。

java"># 默认情况为 ASYNC_FLUSH
flushDiskType = SYNC_FLUSH

2、同步机制：同步复制

为了保证消息不丢失，RocketMQ肯定要通过集群方式进行部署，Broker 通常采用一主多从部署方式，并且采用主从同步的方式做数据复制。

默认方式下，Broker 在接收消息后，写入 master 成功，就可以返回确认响应给生产者了，接着消息将会异步复制到 slave 节点。但是如果这个过程中，Master 的磁盘损坏了。那就会导致数据丢失了。

如果想要解决这个问题，可以配置同步复制的方式，即 Master 在将数据同步到 Slave 节点后，再返回给生产者确认结果。

java"># 默认为 ASYNC_MASTER
brokerRole = SYNC_MASTER

消费者（Consumer）

1、At least Once 机制（先业务后ack）

At least Once：
Consumer 先 pull【主动拉取Broker中的信息】消息到本地，消费完成后，才向服务器返回 ack。

通常消费消息的 ack 机制一般分为两种思路：

第一种：先提交后消费；
第二种：先消费，消费成功后再提交【这个更稳当】；

思路一可以解决重复消费的问题但是会丢失消息，因此 Rocketmq 默认实现的是思路二，由各自 consumer 业务方保证幂等来解决重复消费问题。
幂等：通过给每个消息携带一个唯一标识信息，去数据库进行判断。或者在 producer 的时候就存储一个唯一标识（消息），消费成功删除redis中的消息确保不被重复消费

简单来说就是，在消费者的代码中，一定要在业务逻辑的最后一步returnConsumeConcurrentlyStatuS.CONSUME SUCCESS;
当然，也可以先把数据保存在数据库中，就返回，然后自己再慢慢处理。