赞
踩
最近有很多小伙伴开始找工作,在面试时,面试官经常会问我们这样一个题目:
RabbitMQ如何防止重复消费?
有很多小伙伴这个时候都在想,消息怎么还会重复消费呢???.......
所以他们在面试后就跑来问壹哥,针对这个比较高频的题目,壹哥就在这里为大家来讲讲MQ防止重复消费的实现方案吧。
如果面试官是壹哥的话,那么我想考察的,其实是候选人对技术的基本使用,以及在各种实际应用场景中对可能发生的问题的实际处理能力。所以这道题的考点,最起码要考虑两点:
第一,RabbitMQ中消息的重复消费是如何产生的?我们首先要发现问题,知道问题产生原因;
第二,针对重复消费问题的处理方案及解决机制。
接下来壹哥就根据上述考点,带大家来一起分析这个问题的解题思路。
根据上图,壹哥给大家梳理总结出了消息可能出现重复消费的产生过程,如下:
1. 消费方的业务项目从MQ队列中接收数据;
2. 接着处理业务;
3. 业务处理成功后,消费方项目给MQ返回ack进行手动确认;
4. 返回回调执行结果的过程中,因为网络抖动等原因,回调数据时,MQ没有返回成功。所以MQ队列中的数据会再次发给业务项目,造成重复消费。
针对消息的重复消费问题,壹哥根据上图总结的解决思路如下:
1. 监听器接收MQ队列中的数据;
2. 利用redis的setnx命令,以消息唯一id为key,以消息内容为value,超时时间设置为10秒,存入redis中;
3. 如果能够成功存入,说明没有重复消费,则处理业务,处理完业务后返回ack或者nack确认;
4. 如果存不进去,则说明重复消费,直接返回ack确认的回调信息就可以了。
3.1 发送方测试代码
- /**
- * 测试发送
- * @author 千锋壹哥
- */
- @SpringBootTest(classes = ProducerApplication.class)
- @RunWith(SpringRunner.class)
- public class TestProducer {
-
- @Autowired
- private RabbitTemplate rabbitTemplate;
-
- @Test
- public void contextLoads() throws IOException {
- //给消息封装一个唯一id对象
- CorrelationData messageId = new CorrelationData(UUID.randomUUID().toString());
- //第四个参数: 设置消息唯一id
- rabbitTemplate.convertAndSend("交换器名字","路由键","千锋壹哥测试MQ重复消费处理!!",messageId);
- }
- }
3.2 接收方测试代码
- package com.qf.rabbitmq.topic;
-
- import com.rabbitmq.client.Channel;
- import org.springframework.amqp.core.Message;
- import org.springframework.amqp.rabbit.annotation.RabbitListener;
- import org.springframework.beans.factory.annotation.Autowired;
- import org.springframework.data.redis.core.StringRedisTemplate;
- import org.springframework.stereotype.Component;
-
- import java.io.IOException;
- import java.util.concurrent.TimeUnit;
-
- /**
- * @author 千锋壹哥
- */
- @Component
- public class Consumer {
-
- @Autowired
- private StringRedisTemplate redisTemplate;
-
- @RabbitListener(queues = "队列名字")
- public void getMessage(String msg, Channel channel, Message message) throws IOException {
- //0. 获取MessageId, 消息唯一id
- String messageId = (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation");
- //1. 设置key到Redis
- if(redisTemplate.opsForValue().setIfAbsent(messageId,"0", 10, TimeUnit.SECONDS)) {
-
- //2. 消费消息
- System.out.println("接收到消息:" + msg);
-
- //3. 设置key的value为1
- redisTemplate.opsForValue().set(messageId,"1",10,TimeUnit.SECONDS);
-
- //4. 手动ack
- channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
- }else {
- //5. 获取Redis中的value即可 如果是1,手动ack
- if("1".equalsIgnoreCase(redisTemplate.opsForValue().get(messageId))){
- channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
- }
- }
- }
- }
经过上面的分析,最后壹哥再给大家总结一下这个问题的完整答案。
因为消费方和MQ服务器网络闪断等原因,造成了接收方接受消息后,返回给MQ服务器一个ack确认消息,但MQ却没有接收到,这就造成了重复消费。
利用redis的setnx命令,将消费的消息id存入到redis,超时时间设置为10秒,然后再给mq返回ack。消费前要判断redis中是否存在这个消息id,如果不存在说明没有消费过,则正常消费;如果redis中存在这个消息id,则说明产生了重复消费,此时直接返回ack,不重复执行业务。
如果你想知道更详细的解答,可以参考壹哥的高薪面试题精讲专栏哦:
高薪程序员&面试题精讲系列125之RabbitMQ怎样保证消息的可靠性、不重复及不丢失?
以上就是MQ中,消息重复消费的产生原因及解决思路和对应案例,现在你知道该怎么解决了吗?如果你还有其他疑问,可以在评论区给我们留言哦。关注壹哥,收获多多!
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。