目录

RabbitMQ之常见问题汇总

1.rabbitmq丢消息场景

1.1 消息未持久化丢失

1.2 消费时消息丢失

1.3 如何阻止消息丢失

2. mq消费消息是pull 还是 push

2.1 pull形式消费

2.2 push形式消费

3. mq重复消费场景

3.1 生产端重复情况

3.2 消费端重复

3.3 如何防止

4.prefetch作用

---

RabbitMQ之常见问题汇总

1.rabbitmq丢消息场景

在前面的文章中，我们介绍了mq如何防消息丢失，从消息从生产者发送到Broker的Exchange，再到Queue，再Deliver到消费者，各个环节都有可能会丢失消息。

这里我们主要模拟一下两个场景：消息未持久化丢失和消费时消息丢失。

1.1 消息未持久化丢失

生产者发送了一条未持久化消息，此时消息未被消费，那这条消息存储在内存中

此时重启MQ，那这条消息就会丢失掉。 

1.2 消费时消息丢失

若消费时使用自动确认机制，产生了2条消息，第一条消息由于处理时间比较长，第二条消息已经拉取到了本地BlockingQueue中，再第一条消息处理的过程中，系统宕机了，第二条消息还没来的及消费，就丢失了。

1.3 如何阻止消息丢失

参考 RabbitMQ初步到精通-第五章-RabbitMQ之消息防丢失_Mr-昊哥的博客-CSDN博客

2. mq消费消息是pull 还是 push

mq消费既支持pull的形式也支持push的形式。pull形式，是靠客户端一直拉取服务端的消息，而客户端并不知道什么时候会有消息进入到队列中，会一直取消息，造成额外的系统开销，效率也不高。

而push形式，是一般我们采用的方式。将监听函数注册到服务中，mqBroker收到消息进行投递，再调用监听函数方法消费消息。

2.1 pull形式消费

代码中使用：

        for (int i = 0; i < 2; i++) {GetResponse response = channel.basicGet(QUEUE_NAME, true);System.out.println("消费：" + new String(response.getBody()));}

抓包情况：

都是先发出请求，再将结果返回。

2.2 push形式消费

代码：

        DefaultConsumer consumer = new DefaultConsumer(channel) {@SneakyThrows@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {Thread.sleep(1000);System.out.println("接收到消息:" + new String(body, "UTF-8"));channel.basicAck(envelope.getDeliveryTag(), false);}};channel.basicConsume(QUEUE_NAME, false, consumer);

抓包情况：

服务端将消息推送过来：

3. mq重复消费场景

3.1 生产端重复情况

从生产端即产生了重复的情况，多由于系统bug，或系统间调用造成重试等原因，推出了重复的消息。

3.2 消费端重复

消费使用手动ACK模式，两个消费者，消费者1，虽然消费到了消息，由于处理时间长，未能及时ACK，这时候系统宕机，mq会将NACK状态的消息变为Ready状态，又推送给了消费者2 ，这样就造成了，同一条消息 同时在消费者1和消费者2都消费到了。

生产者：

public class RepeatProducer {public static final String QUEUE_NAME = "work";//生产者public static void main(String[] args) throws Exception {//1、获取connectionConnection connection = RabbitCommonConfig.getConnection();//2、创建channelChannel channel = connection.createChannel();for (int i = 1; i <= 3; i++) {sendMsg(channel, i);Thread.sleep(1000);}//4、关闭管道和连接channel.close();connection.close();}private static void sendMsg(Channel channel, int k) throws IOException {//3、发送消息到exchangeString msg = "hello work :" + k;channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());System.out.println("生产者发布消息成功！" + k);}}

消费者1：

public class RepeatConsumer1 {public static final String QUEUE_NAME = "work";//消费者public static void main(String[] args) throws Exception {//1、获取连对象、Connection connection = RabbitCommonConfig.getConnection();//2、创建channelChannel channel = connection.createChannel();channel.basicQos(1);//3、创建队列channel.queueDeclare(QUEUE_NAME, true, false, false, null);//4.开启监听QueueDefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("Consumer1 接收到消息:" + new String(body, "UTF-8"));try {Thread.sleep(1000 * 60);} catch (InterruptedException e) {e.printStackTrace();}//手动ACK（接收信息，指定是否批量操作）channel.basicAck(envelope.getDeliveryTag(), false);}};//5.关闭自动ACKchannel.basicConsume("work", false, consumer);System.out.println("消费者1开始监听队列");//6、键盘录入，让程序不结束！System.in.read();//7、释放资源channel.close();connection.close();}}

消费者2：

public class RepeatConsumer2 {public static final String QUEUE_NAME = "work";//消费者public static void main(String[] args) throws Exception {//1、获取连对象、Connection connection = RabbitCommonConfig.getConnection();//2、创建channelChannel channel = connection.createChannel();channel.basicQos(1);//3、创建队列channel.queueDeclare(QUEUE_NAME, true, false, false, null);//4.开启监听QueueDefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("Consumer2 接收到消息:" + new String(body, "UTF-8"));try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}//手动ACK（接收信息，指定是否批量操作）channel.basicAck(envelope.getDeliveryTag(), false);}};//5.关闭自动ACKchannel.basicConsume(QUEUE_NAME, false, consumer);System.out.println("消费者2开始监听队列");//6、键盘录入，让程序不结束！System.in.read();//7、释放资源channel.close();connection.close();}}

结果：

第一次运行：

消费者1开始监听队列
Consumer1 接收到消息:hello work :1

消费者2开始监听队列
Consumer2 接收到消息:hello work :2
Consumer2 接收到消息:hello work :3

运行过程中停掉消费者1

消费者2开始监听队列
Consumer2 接收到消息:hello work :2
Consumer2 接收到消息:hello work :3
Consumer2 接收到消息:hello work :1

3.3 如何防止

消费时引入幂等机制，使用分布式锁，数据库唯一索引等控制。

4.prefetch作用

接触到prefetch分别会在java和spring 客户端中出现，只有需要手动ACK的时候才起作用。

设置了此值，会控制mq服务端给客户端推送的消息数量。

例如设置15，mq堆积了大量消息的情况下，会首次推送给 客户端15个消息，若消息消费慢的情况，mq的服务端会始终保证15个unack的消息的前提下，给客户端推送。

1.java amqp client

        channel.basicQos(15);//开启手动确认channel.basicConsume(QUEUE_NAME, false, consumer);

2. spring 配置中

spring.rabbitmq.listener.simple.acknowledge-mode = manual
spring.rabbitmq.listener.simple.prefetch = 15

spring若没进行配置默认250

private volatile int prefetchCount = DEFAULT_PREFETCH_COUNT;
public static final int DEFAULT_PREFETCH_COUNT = 250;