rocketmq 消息丢失总结

Dear 丶 2022-09-10 14:17 453阅读 0赞

总结rocketmq消息丢失四种情况

1、 生产者网络波动 消息丢失

解决方法：利用rocketmq自带的事务机制处理，首先发送half消息于rocketmq服务器，此时消息于消费者不可见，等生产者业务处理完成则发送，否则回滚，如果half消息发送成功，且RocketMQ返回成功响应，则消费者正常消费，否则回滚

核心代码展示

生产者  
//设置回调函数  
producer.setTransactionCheckListener(transactionCheckListener);  
//发送事务消息  
SendResult sendResult = producer.sendMessageInTransaction(msg, tranExecuter, null)

回调函数  
回调方法实现 TransactionListener  
//重写执行事务方法 （生产者主动告诉rocketmq状态）  
LocalTransactionState executeLocalTransaction(Message message, Object o)  
根据事务执行情况返回相应的LocalTransactionState

//重写rocketmq回查事务方法 （rocketmq主动向消费者回查状态）  
LocalTransactionState checkLocalTransaction(MessageExt messageExt)  
根据事务执行情况返回相应的LocalTransactionState

2、rocketmq写入内存队列，却没来得及进行磁盘持久化

要保证消息不丢失，首先需要将os cache的异步刷盘策略改为同步刷盘，这一步需要修改Broker的配置文件，将flushDiskType改为SYNC\_FLUSH同步刷盘策略，默认的是ASYNC\_FLUSH异步刷盘。

3、消息写入磁盘，但是磁盘损坏  
需要对RocketMQ采用主从机构，集群部署，Leader中的数据在多个Follower中都存有备份，防止单点故障

4、消费者未消费完就通知rocketmq消费成功  
需要在RocketMQ在消费者中注册了一个监听器，当消费者获取到了消息，就会去回调这个监听器函数，去处理里面的消息当你的消息处理完毕之后，才会返回

核心代码展示

//注册消息监听器处理消息  
consumer.registerMessageListener(new MessageListenerConcurrently() \{  
@Override  
public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context)\{  
//对消息进行处理  
return ConsumeConcurrentlyStatus.CONSUME\_SUCCESS;  
\}  
\});

rocketmq 有一个至少成功一次的机制(At least Once)  
如果消费者消费失败，会就行重试，默认16次重试机会，直至消费者消费成功为止，否则加入死信队列

死信队列  
死信队列消息的前提是基于consumer消费消息异常后经过多次重试队列的投递依赖消费失败最终会进入死信队列。

重投处理逻辑先将消息投递到重试Topic的%RETRY%consumerGroup队列，超过最大重试次数后将消息投递到死信Topic的%DLQ%consumerGroup队列。

注意：重试可能会产生出现消息重复消费的事件，通过保证业务幂等性解决