消费者预取

概述

消费者预取是通道预取机制的一种扩展。

AMQP 0-9-1 规范规定了 basic.qos 方法，旨在限制消费时通道（或连接）上未确认消息的数量（即“预取计数”）。遗憾的是，通道并不是实现此功能的理想作用域——由于单个通道可能从多个队列进行消费，因此通道和队列需要在发送每条消息时相互协调，以确保不超过限制。这在单台机器上运行缓慢，而在跨集群消费时则非常缓慢。

此外，对于许多用例而言，指定应用于每个消费者的预取计数更为自然。

因此，当涉及如何将预取应用于通道上的多个消费者时，RabbitMQ 与 AMQP 0-9-1 规范略有不同。

AMQP 0-9-1 中 prefetch_count 的含义	RabbitMQ 中 prefetch_count 的含义
在通道的所有消费者之间共享	分别应用于通道上的每个新消费者

单个消费者

以下 Java 中的基础示例将一次接收最多 10 条未确认的消息

Channel channel = ...;
Consumer consumer = ...;
channel.basicQos(10); // Per consumer limit
channel.basicConsume("my-queue", false, consumer);

值 0 被视为无限，允许任意数量的未确认消息。

Channel channel = ...;
Consumer consumer = ...;
channel.basicQos(0); // No limit for this consumer
channel.basicConsume("my-queue", false, consumer);

独立消费者

此示例在同一通道上启动两个消费者，每个消费者将独立地一次接收最多 10 条未确认的消息

Channel channel = ...;
Consumer consumer1 = ...;
Consumer consumer2 = ...;
channel.basicQos(10); // Per consumer limit
channel.basicConsume("my-queue1", false, consumer1);
channel.basicConsume("my-queue2", false, consumer2);

多个消费者共享限制

AMQP 0-9-1 规范并未说明如果您多次使用不同的 global 值调用 basic.qos 会发生什么。RabbitMQ 将其解释为应独立强制执行这两个预取限制；只有当未确认消息的两个限制都未达到时，消费者才会接收新消息。

例如：

Channel channel = ...;
Consumer consumer1 = ...;
Consumer consumer2 = ...;
channel.basicQos(10, false); // Per consumer limit
channel.basicQos(15, true);  // Per channel limit
channel.basicConsume("my-queue1", false, consumer1);
channel.basicConsume("my-queue2", false, consumer2);

这两个消费者之间总共最多只会拥有 15 条未确认的消息，且每个消费者最多拥有 10 条消息。由于在通道和队列之间进行协调以强制执行全局限制会产生额外的开销，因此这将比上述示例更慢。

可配置的默认预取

RabbitMQ 可以使用默认的预取值，如果消费者未指定预取值，则会应用该值。该值可以在高级配置文件中配置为 rabbit.default_consumer_prefetch。

%% advanced.config file
[
 {rabbit, [
       {default_consumer_prefetch, {false,250}}
     ]
 }
].