RabbitMQ初步到精通-第十一章-RabbitMQ之常见问题汇总

目录

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
            @Override
            public 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、获取connection
        Connection connection = RabbitCommonConfig.getConnection();
        //2、创建channel
        Channel 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、发送消息到exchange
        String 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、创建channel
        Channel channel = connection.createChannel();
 
        channel.basicQos(1);
        //3、创建队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
 
        //4.开启监听Queue
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public 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.关闭自动ACK
        channel.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、创建channel
        Channel channel = connection.createChannel();
 
        channel.basicQos(1);
        //3、创建队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
 
        //4.开启监听Queue
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public 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.关闭自动ACK
        channel.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;