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

news/2024/4/16 20:48:09/文章来源:https://blog.csdn.net/syc000666/article/details/129236542

一. 前言

最近有很多小伙伴开始找工作，在面试时，面试官经常会问我们这样一个题目：

RabbitMQ如何防止重复消费?

有很多小伙伴这个时候都在想，消息怎么还会重复消费呢???.......

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

二. 面试题考点

如果面试官是壹哥的话，那么我想考察的，其实是候选人对技术的基本使用，以及在各种实际应用场景中对可能发生的问题的实际处理能力。所以这道题的考点，最起码要考虑两点:

第一，RabbitMQ中消息的重复消费是如何产生的？我们首先要发现问题,知道问题产生原因；
第二，针对重复消费问题的处理方案及解决机制。

三. 解题分析

接下来壹哥就根据上述考点，带大家来一起分析这个问题的解题思路。

1. RabbitMQ消息重复消费的产生原因

根据上图，壹哥给大家梳理总结出了消息可能出现重复消费的产生过程，如下：

1. 消费方的业务项目从MQ队列中接收数据；
2. 接着处理业务；
3. 业务处理成功后，消费方项目给MQ返回ack进行手动确认；
4. 返回回调执行结果的过程中，因为网络抖动等原因，回调数据时，MQ没有返回成功。所以MQ队列中的数据会再次发给业务项目，造成重复消费。

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

针对消息的重复消费问题，壹哥根据上图总结的解决思路如下：

1. 监听器接收MQ队列中的数据；
2. 利用redis的setnx命令，以消息唯一id为key，以消息内容为value，超时时间设置为10秒，存入redis中；
3. 如果能够成功存入，说明没有重复消费，则处理业务，处理完业务后返回ack或者nack确认；
4. 如果存不进去，则说明重复消费，直接返回ack确认的回调信息就可以了。

3. 解决重复消费的案例代码

3.1 发送方测试代码

/*** 测试发送* @author 千锋壹哥*/
@SpringBootTest(classes = ProducerApplication.class)
@RunWith(SpringRunner.class)
public class TestProducer {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void contextLoads() throws IOException {//给消息封装一个唯一id对象CorrelationData messageId = new CorrelationData(UUID.randomUUID().toString());//第四个参数: 设置消息唯一idrabbitTemplate.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 {@Autowiredprivate StringRedisTemplate redisTemplate;@RabbitListener(queues = "队列名字")public void getMessage(String msg, Channel channel, Message message) throws IOException {//0. 获取MessageId, 消息唯一idString messageId = (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation");//1. 设置key到Redisif(redisTemplate.opsForValue().setIfAbsent(messageId,"0", 10, TimeUnit.SECONDS)) {//2. 消费消息System.out.println("接收到消息：" + msg);//3. 设置key的value为1redisTemplate.opsForValue().set(messageId,"1",10,TimeUnit.SECONDS);//4.  手动ackchannel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}else {//5. 获取Redis中的value即可 如果是1，手动ackif("1".equalsIgnoreCase(redisTemplate.opsForValue().get(messageId))){channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}}}
}