0. 引言

在金融、电商等对数据完整性要求极高的行业，消息的丢失可能会导致数据不一致，严重影响业务逻辑和数据统计，也影响客户体验，所以在很多业务场景下，我们都要求数据不能丢失。而rocketmq中，如何对消息防丢失进行处理的呢？

1. 原理

1.2 产生消息丢失的场景

首先我们要理解消息的传递过程，在哪些阶段会导致消息丢失，才能知道如何进行防控。

我们之前分析过rabbitmq如何保证消息不丢失, rabbitmq内部有交换机这一转发步骤，所以相对比rocketmq更加复杂，但是两者的分析方法是一致的。

rocketmq的消息传递分为3个阶段：
（1）生产者发送消息到broker的队列中
（2）broker存储消息
（3）消费者到队列获取消息进行消费

而这三个阶段可能会导致消息丢失的场景是什么呢？其实由rabbitmq的分析我们可以得到启发。
（1）生产者发送消息到broker的队列中

生产者在发出消息后，可能因为网络异常、broker宕机，导致发出的消息实际并没有到达broker

（2）broker存储消息

broker的存储机制是将消息先存储到内存，存储完成后再发送回执给生产者，然后再异步将数据刷到磁盘，但如果在这个刷盘这个过程中broker宕机了，也会导致消息丢失

（3）消费者到队列获取消息进行消费

broker在将消息发出后，同样可能因为网络异常、消费者宕机或者消息者消费到一半产生错误等因素，导致消息实际并没有被消费者消费，但broker又扣除了这条消息，就会导致消息丢失

1.2 防丢失措施

阶段一：生产者发送消息到broker的队列中

1、因为发送到broker期间网络因素我们很难干预，也很难百分比保证。第一点我们能做的，如果其中一个broker宕机，那能有备用节点顶上，保证可用性。于是第一项就是多节点部署broker

2、但万一节点都挂了呢，或者整个机房网络瘫痪了，如何保证消息不丢失，我们只要从上游控制，如果下游不通时，就不要发了，待会再发。于是也衍生出消息发送失败时的重试机制。

3、但如果一直重发不成功怎么办呢，那就需要下游告知上游，这次发送没成功，你记录好状态，这就是broker要有返回状态告知，否则生产者也不知道到底发送成功没有。broker中提供了3种发送方式：同步、异步、单向（详见之前的文章： RocketMQ快速入门：集成java客户端实现各类消息发送｜异步、同步、顺序、单向、延迟、事务（五）附带源码）。

这三种方式中单向肯定不行，他是不管返回结果的，最容易丢失消息。而异步需要设置回调函数，在回调函数中处理发送失败时的逻辑，如果对于一些场景回调里很能补救，最常见的就是回调里进行重发，所以最优先保证消息可靠的就是同步发送的方式，一旦获取到发送失败，就进行补救处理，或者不再继续后续的业务逻辑，整个流程直接报错打回

总结一下，生产阶段保证消息可靠的手段包括多节点部署broker, 消息重发、同步发送。这几种方式实际上是可以配合使用的， 比如多节点部署，通过同步发送，发送失败时进行3次重发，都重发失败则记录状态。

阶段二：broker存储消息

存储阶段导致丢失的原因就是因为broker默认的是异步刷盘机制，如果改成同步刷盘呢，先存储到内存，然后刷新到磁盘，刷新成功后才给生产者返回成功收到的回执，以此保证消息可靠。rocketmq中也是支持同步刷盘的。

但如果只有一个节点的话，即使同步刷盘，当broker宕机后，没有备用的，还是会导致服务不可用，相对可靠性就没有保障了。所以同步刷盘，也可以配合着多节点部署使用。

当然如果你的场景对可用性要求不高，即使宕机，只要报错会生产者就行，那同步刷盘也足够了

总结一下，存储阶段主要依赖同步刷盘和多节点部署来保障可靠性，当然多节点部署可以根据业务情况和成本预算选择。

阶段三：消费者到队列获取消息进行消费

消费阶段的丢失可能性主要来源于消费者没处理好之前就宕机或则异常了，首先一点能想到的那就多个消费者呢，但实际上多点部署在消费阶段并不能解决问题，因为rocketmq消费模式有广播模式和集群模式，广播模式下每个节点都会收到消息，这个模式下的天然就是多节点部署。而集群模式本身也是基于多消费者的情况，但两者都无法保障当消息发送给某一节点后，这个节点拉去了消息，但没实际处理完就异常的场景。

所以就考虑当消费者消费完成后，再给broker发送成功消费的回执，这时broker才更新消息偏移量，将消息标识为被消费。如此才能保障消息的可靠性。

2. 实现

2.1 生产阶段

1、多节点部署，可以部署主从或集群模式，这里不讲解详细的搭建流程，后续单独讲解

2、同步发送，主要通过producer.send来实现同步发送

public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {// 声明groupDefaultMQProducer producer = new DefaultMQProducer("group_test");// 声明namesrv地址producer.setNamesrvAddr("localhost:9876");// 设置重试次数producer.setRetryTimesWhenSendFailed(2);// 启动实例producer.start();// 设置消息的topic,tag以及消息体Message msg = new Message("topic_test", "tag_test", "消息内容1".getBytes(StandardCharsets.UTF_8));// 发送消息，并设置10s连接超时SendResult send = producer.send(msg, 10000);System.out.println("发送结果："+send);// 关闭实例producer.shutdown();}

如果是springboot集成的，可以通过SendResult sendResult = rocketMQTemplate.syncSend("topic_test:tag_test", message);来实现。通过发送结果SendResult对象，来判断发送失败后的处理逻辑

3、发送重试
（1）首先手动重发的实现很简单，只需要根据send.getSendStatus()状态来判断，如果需要重发多次的，可以结合guava-retry 等重发组件来更方便的实现

// 发送消息，并设置10s连接超时SendResult send = producer.send(msg, 10000);System.out.println("发送结果："+send);if(!send.getSendStatus().equals(SendStatus.SEND_OK)){// 发送失败，手动重发send = producer.send(msg, 10000);}

（2）当然rocketmq也封装好了重试机制给我们使用，其重试机制采用衰减的形式，也就是重试间隔时间会逐渐增加

我们只需要通过producer.setRetryTimesWhenSendFailed(2);方法即可设置，会在发送失败时自动触发重新发送，同时如果超过设置的超时时间还未接收到成功的结果也会触发重发机制，就不需要我们手动书写重发逻辑了，更加推荐这种方式。

        // 声明groupDefaultMQProducer producer = new DefaultMQProducer("group_test");// 声明namesrv地址producer.setNamesrvAddr("localhost:9876");// 设置重试次数producer.setRetryTimesWhenSendFailed(2);// 启动实例producer.start();

2.2 存储阶段

1、主要将broker的刷盘策略设置为同步刷盘，需要修改broker.conf配置文件

# 设置为同步刷盘模式
flushDiskType = SYNC_FLUSH

2、如果配置的是多节点，一般是主从模式，为了防止主节点有数据，从节点没刷到数据的情况，就需要开启从节点刷盘后再返回ACK回执给生产者，需要修改从节点broker配置文件

# 默认为 ASYNC_MASTER
brokerRole=SYNC_MASTER

broker提供了两种主从同步模式：ASYNC_MASTER异步 和 SYNC_MASTER同步

• ASYNC_MASTER：

消息发送到master节点后，开启一个异步线程更新给从节点，这个过程中有消息同步丢失的风险，优点是性能高

• SYNC_MASTER：

消息发送到master节点后，同步更新到从节点，当从节点更新完再返回成功的ACK回执给生产者，表示消息发送成功，可靠性高，但性能会有所下降

3、关于多节点部署，我们在后续单独讲解

2.3 消费阶段

1、其实现就是消费后返回成功状态即可

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group_test");consumer.setNamesrvAddr("127.0.0.1:9876");// 集群消费模式consumer.setMessageModel(MessageModel.CLUSTERING);// 设置topicconsumer.subscribe("topic_test", "*");// 设置重试次数consumer.setMaxReconsumeTimes(2);// 注册回调函数，处理消息consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {for (MessageExt msg : list) {String topic = msg.getTopic();try {String messageBody = new String(msg.getBody(), "utf-8");System.out.println(topic+":"+messageBody);} catch (UnsupportedEncodingException e) {e.printStackTrace();}}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 启动消费者实例consumer.start();

2、如果返回的是ConsumeConcurrentlyStatus.RECONSUME_LATER状态，消费者就会触发稍后重试机制进行重新消费，同样的可以通过consumer.setMaxReconsumeTimes设置最大重试次数

// 设置重试次数
consumer.setMaxReconsumeTimes(2);

其重试次数和时延等级与生产重试是一致的