高频面试题|RabbitMQ如何防止重复消费？

一. 前言

最近有很多小伙伴开始找工作&＃xff0c;在面试时&＃xff0c;面试官经常问到一个题目&＃xff1a;RabbitMQ如何防止重复消费?

有很多小伙伴这个时候都在想&＃xff0c;消息怎么就会重复消费呢???.......

所以他们在面试后就跑来问健哥&＃xff0c;针对这个比较高频的题目&＃xff0c;健哥就在这里为大家来讲讲MQ防止重复消费的实现方案吧。

二. 面试题考点

如果面试官是健哥的话&＃xff0c;那么我想考察的&＃xff0c;其实就是候选人除了对技术的基本使用之外&＃xff0c;再就是在各种实际应用场景中对可能发生问题的实际处理能力。

所以这道题的考点&＃xff0c;最起码有两点:

• 第一是RabbitMQ中消息的重复消费是如何产生的&＃xff0c;我们首先要发现问题,&＃xff0c;知道问题产生原因&＃xff1b;

• 第二是针对这个重复消费问题的处理方案及机制。

三. 解题分析

接下来健哥就根据上述考点&＃xff0c;带大家来一起分析这个问题的解题思路。

3.1RabbitMQ消息重复消费的产生原因

根据上图&＃xff0c;健哥给大家梳理总结出了消息重复消费的产生过程&＃xff0c;如下&＃xff1a;

1. 消费方的业务项目从MQ队列中接收数据&＃xff1b;

2. 接着处理业务&＃xff1b;

3. 业务处理成功后&＃xff0c;消费方项目给MQ返回ack进行手动确认&＃xff1b;

4. 返回回调执行结果的过程中&＃xff0c;因为网络抖动等原因&＃xff0c;回调数据时&＃xff0c;MQ没有返回成功&＃xff0c;所以MQ队列中的数据会再次发给业务项目&＃xff0c;造成重复消费。

3.2. RabbitMQ消息重复消费的处理方案

针对消息的重复消费问题&＃xff0c;健哥根据上图总结的解决思路如下&＃xff1a;

1. 监听器接收MQ队列中的数据&＃xff1b;

2. 利用redis的setnx命令&＃xff0c;以消息唯一id为key&＃xff0c;以消息内容为value&＃xff0c;超时时间设置为10秒&＃xff0c;存入redis中&＃xff1b;

3. 如果能够成功存入&＃xff0c;说明没有重复消费&＃xff0c;则处理业务&＃xff0c;处理完业务后返回ack或者nack确认&＃xff1b;

4. 如果存不进去&＃xff0c;则说明重复消费&＃xff0c;直接返回ack确认的回调信息就可以了。

3.3解决重复消费的案例代码

• 发送方测试代码

  /*** 测试发送* &＃64;author 千锋健哥*/
&＃64;SpringBootTest(classes &＃61; ProducerApplication.class)
&＃64;RunWith(SpringRunner.class)
public class TestProducer {&＃64;Autowiredprivate RabbitTemplate rabbitTemplate;&＃64;Testpublic void contextLoads() throws IOException {//给消息封装一个唯一id对象CorrelationData messageId &＃61; new CorrelationData(UUID.randomUUID().toString());//第四个参数: 设置消息唯一idrabbitTemplate.convertAndSend("交换器名字","路由键","千锋健哥测试MQ重复消费处理&＃xff01;&＃xff01;",messageId);}
}


  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;/*** &＃64;author 千锋健哥*/
&＃64;Component
public class Consumer {&＃64;Autowiredprivate StringRedisTemplate redisTemplate;&＃64;RabbitListener(queues &＃61; "队列名字")public void getMessage(String msg, Channel channel, Message message) throws IOException {//0. 获取MessageId, 消息唯一idString messageId &＃61; (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation");//1. 设置key到Redisif(redisTemplate.opsForValue().setIfAbsent(messageId,"0", 10, TimeUnit.SECONDS)) {//2. 消费消息System.out.println("接收到消息&＃xff1a;" &＃43; msg);//3. 设置key的value为1redisTemplate.opsForValue().set(messageId,"1",10,TimeUnit.SECONDS);//4.  手动ackchannel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}else {//5. 获取Redis中的value即可 如果是1&＃xff0c;手动ackif("1".equalsIgnoreCase(redisTemplate.opsForValue().get(messageId))){channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}}}
}

  • 接收方测试代码

四. 总结

经过上面的分析&＃xff0c;最后健哥再给大家总结一下这个问题的完整答案。

• 问题产生原因&＃xff1a;

因为消费方和MQ服务器网络闪断等原因&＃xff0c;造成了接收方消费后&＃xff0c;返回给MQ服务器一个ack确认消息&＃xff0c;结果MQ没有接收到&＃xff0c;造成了重复消费。

• 解决过程&＃xff1a;

利用redis的setnx命令&＃xff0c;将消费的消息id存入到redis&＃xff0c;超时时间设置为10秒&＃xff0c;然后再给mq返回ack。消费前要判断redis中是否存在这个消息id&＃xff0c;如果不存在说明没有消费过&＃xff0c;则正常消费&＃xff1b;如果redis中存在这个消息id&＃xff0c;则说明重复消费&＃xff0c;直接返回ack&＃xff0c;不重复执行业务。

以上就是MQ中消息重复消费的产生原因及解决思路和对应案例&＃xff0c;现在你知道该怎么解决了吗&＃xff1f;