RabbitMQ消费者预取机制与消息确认机制

什么是消费者预取

消费者预取是对AMQP协议中通道预取机制的扩展。简单来说，它控制着消费者一次能接收多少条未确认的消息。

image-20210710-1

与AMQP标准的区别

按照AMQP 0-9-1标准：
● 预取数量是在通道级别共享的（同一个通道上所有消费者共用这个数量）
但RabbitMQ做了改进：
● 预取数量是针对每个消费者单独计算的（每个消费者有自己的限额）
RabbitMQ这样做有两个原因：

  1. 标准方式效率低 - 当单个通道从多个队列消费时，需要频繁协调
  2. 实际使用时，为每个消费者单独设置限额更符合直觉

使用示例

单个消费者案例

1
2
3
4
Channel channel = ...;
Consumer consumer = ...;
channel.basicQos(10); // 这个消费者最多同时处理10条未确认消息
channel.basicConsume("my-queue", false, consumer);

如果设置为0，表示不限制数量：

1
channel.basicQos(0); // 这个消费者可以处理无限数量的消息

多个独立消费者

同一个通道上的两个消费者，各自有10条的限额：

1
2
3
4
5
6
Channel channel = ...;
Consumer consumer1 = ...;
Consumer consumer2 = ...;
channel.basicQos(10); // 每个消费者限额10条
channel.basicConsume("my-queue1", false, consumer1);
channel.basicConsume("my-queue2", false, consumer2);

共享限额的消费者

也可以设置两种限额：每个消费者的限额 + 整个通道的总限额

1
2
channel.basicQos(10, false); // 每个消费者最多10条
channel.basicQos(15, true);  // 整个通道最多15条

这样两个消费者合计最多15条未确认消息，且每个消费者不超过10条。不过这种模式性能会差一些。

默认预取设置

RabbitMQ允许配置默认的预取值（当消费者没有明确设置时使用）。可以在配置文件中设置：

1
2
3
4
5
6
[
 {rabbit, [
       {default_consumer_prefetch, {false,250}}  // 默认每个消费者250条
     ]
 }
]

总结

RabbitMQ的预取机制比AMQP标准更灵活实用，可以：
● 为每个消费者单独设置处理上限
● 也可以设置通道级别的总上限
● 支持无限处理模式（设置为0时）
● 可以配置默认值简化代码

Consumer消息确认机制

在RabbitMQ中，消费者确认机制（Consumer Delivery Acknowledgements）是确保消息可靠传递的核心环节，它通过一套精细的控制机制来协调消息的安全处理与系统吞吐量之间的平衡。每个被投递的消息都会被分配一个唯一的交付标识符（Delivery Tag），这是一个单调递增的整数值，在通道（Channel）范围内有效，消费者通过此标识符对特定消息进行确认操作。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
package com.lixiang.rabbitmq;

import com.lixiang.rabbitmq.utils.RabbitConstant;
import com.lixiang.rabbitmq.utils.RabbitUtils;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

@Slf4j
public class SmsSenderHigh {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        final Channel channel = connection.createChannel();
        
        channel.basicQos(5);
        channel.queueDeclare(RabbitConstant.QUEUE_SMS, true, false, false, null);

        channel.basicConsume(RabbitConstant.QUEUE_SMS, false, null, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                    byte[] body) throws IOException {
                String jsonSMS = new String(body);
                log.info("SMSSender-短信发送成功:{}", jsonSMS);
                try {
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }
}

Java客户端中，当消费者处理完消息后，可调用channel.basicAck(deliveryTag, false)进行单条消息确认，或使用channel.basicAck(deliveryTag, true)批量确认所有小于等于该标签的消息，这种批量确认机制能显著提升处理效率，尤其在消息密集场景下。
RabbitMQ提供两种消费者确认模式（Consumer Acknowledgement Modes）：手动模式（显式调用ACK/NACK）和自动模式（自动确认）。数据安全性要求高的场景必须采用手动模式，以避免消息在消费者崩溃时丢失。当消息处理失败时，可通过channel.basicNack(deliveryTag, false, true)进行否定确认，第三个参数控制是否重新入队（requeue），此时消息会回到队列头部，可能被原消费者重复获取。若希望将消息路由到死信队列（DLX），则需设置requeue=false并预先配置DLX策略。
消费者预取数量（Prefetch Count）的设置直接影响系统吞吐量与公平性。通过channel.basicQos(10)限制未确认消息的最大数量，可防止单个消费者独占队列导致负载不均，但过小的预取值会降低处理速度，而过大的预取值可能引发消费者内存溢出。当消费者异常断开时，所有未确认的消息会自动重新入队（Automatic Requeueing），这是通过RabbitMQ的连接心跳检测机制实现的。值得注意的是，重新入队的消息可能因网络延迟出现乱序情况，业务逻辑需考虑幂等性设计。