RocketMQ—RocketMQ消息重复消费问题
重复消费问题的描述
什么情况下会发生重复消费的问题:
-
生产者多次投递消息:如果生产者发送消息时,连接有延迟,MQ还没收到消息,生产者又发送了一次消息;
-
消费者方扩容时会重试:
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;