RocketMQ—RocketMQ消息重复消费问题

RocketMQ—RocketMQ消息重复消费问题

重复消费问题的描述

什么情况下会发生重复消费的问题：

1. 生产者多次投递消息：如果生产者发送消息时，连接有延迟，MQ还没收到消息，生产者又发送了一次消息；

2. 消费者方扩容时会重试：

   topic中有四个queue，queue2中有一个消息MSG，如果此时只有一个消费者，MSG就会被被consumer1获取到；

   

   但是consumer1获取到消息时，位点还没来得及前移，又有了一个消费者Consumer2，这样就会产生重平衡；

   

   Consumer1消费0和1队列，Consumer2消费2和3队列，因为上一步的消息还没消费成功，所以会给Consumer2一份消息，这样就会产生重复消费的问题。

重复消费问题的解决方案

一般情况下，我们会在消费者端解决重复消费的问题，消费者端需要进行去重，去重关键点就是要找到消息的唯一标记，所以我们在发送消息时会带有一个key，消费者拿到相同的key就不进行操作了。（一般使用redis存储消费过的key，或者使用mysql存储消费记录，把key设置成唯一索引。）

mysql解决方案

简单建一个去重表，有id，key两个字段，key设置唯一索引。

错误做法

if(key存在){
   
	直接返回;
}
执行业务逻辑。
插入key记录到MySQL;

这样做是不正确的，因为不能保证并发安全，如果有两个线程同时进来，这样都会查不到这个key，都会执行消费的逻辑。

正确做法

try{
   
    插入key记录到MySQL;//MySQL唯一索引内部会保证唯一
}catch (SQLException e) {
   
    if (e instanceof SQLIntegrityConstraintViolationException) {
   
        // 唯一索引冲突异常
        // 说明消息来过了
        记录日志;
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
    其他错误依然会报错，会重新投递到消息队列；
}
处理业务逻辑
如果业务报错 则删除掉这个去重表记录
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;