带批处理的 @RabbitListener

Doubao 中英对照 @RabbitListener with Batching

当接收一批消息时，拆批通常由容器执行，监听器一次接收一条消息被调用。从 2.2 版开始，你可以配置监听容器工厂和监听器，使其在一次调用中接收整批消息——只需将工厂的 batchListener 属性设为 true，并让方法的载荷参数为 List 或 Collection：

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory());
    factory.setBatchListener(true);
    return factory;
}

@RabbitListener(queues = "batch.1")
public void listen1(List<Thing> in) {
    ...
}

// or

@RabbitListener(queues = "batch.2")
public void listen2(List<Message<Thing>> in) {
    ...
}

将 batchListener 属性设为 true，会自动关闭工厂所创建容器中的 deBatchingEnabled 容器属性（除非 consumerBatchEnabled 为 true——见下文）。实际上，拆批从容器移到了监听适配器，由适配器创建传递给监听器的列表。

启用批处理的工厂不能与多方法监听器一起使用。

同样从 2.2 版开始，当逐条接收批量消息时，最后一条消息包含一个布尔消息头，其值为 true。你可以通过给监听器方法添加一个 @Header(AmqpHeaders.LAST_IN_BATCH) 布尔类型的 last 参数来获取该消息头。该消息头映射自 MessageProperties.isLastInBatch()。此外，AmqpHeaders.BATCH_SIZE 会在每个消息片段中填充为该批的大小。

此外，SimpleMessageListenerContainer 新增了一个 consumerBatchEnabled 属性。当它为 true 时，容器会创建一批消息，最多达 batchSize；如果 receiveTimeout 超时仍没有新消息到达，则会投递一个部分批次。如果接收到一个由生产者创建的批次，它会被拆批并加入到消费者侧的批次中；因此实际投递的消息数可能超过 batchSize，后者表示从 broker 接收到的消息数。当 consumerBatchEnabled 为 true 时，deBatchingEnabled 必须为 true；容器工厂会强制执行这一要求。

@Bean
public SimpleRabbitListenerContainerFactory consumerBatchContainerFactory() {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(rabbitConnectionFactory());
    factory.setConsumerTagStrategy(consumerTagStrategy());
    factory.setBatchListener(true); // configures a BatchMessageListenerAdapter
    factory.setBatchSize(2);
    factory.setConsumerBatchEnabled(true);
    return factory;
}

将 consumerBatchEnabled 与 @RabbitListener 一起使用时：

@RabbitListener(queues = "batch.1", containerFactory = "consumerBatchContainerFactory")
public void consumerBatch1(List<Message> amqpMessages) {
    ...
}

@RabbitListener(queues = "batch.2", containerFactory = "consumerBatchContainerFactory")
public void consumerBatch2(List<org.springframework.messaging.Message<Invoice>> messages) {
    ...
}

@RabbitListener(queues = "batch.3", containerFactory = "consumerBatchContainerFactory")
public void consumerBatch3(List<Invoice> strings) {
    ...
}

• 第一个以接收到的原始、未转换的 org.springframework.amqp.core.Message 列表调用。
• 第二个以带有已转换载荷和已映射消息头/属性的 org.springframework.messaging.Message<?> 列表调用。
• 第三个以已转换的载荷调用，无法访问消息头/属性。

你还可以添加一个 Channel 参数，这在使用 MANUAL ack 模式时经常用到。在第三个示例中这不太有用，因为你无法访问 delivery_tag 属性。

Spring Boot 为 consumerBatchEnabled 和 batchSize 提供了配置属性，但没有为 batchListener 提供。从 3.0 版开始，在容器工厂上将 consumerBatchEnabled 设为 true 也会把 batchListener 设为 true。当 consumerBatchEnabled 为 true 时，监听器必须是批监听器。

从 3.0 版开始，监听方法可以消费 Collection<?> 或 List<?>。

备注

批模式下的监听器不支持回复，因为批次中的消息与所产生的单个回复之间可能不存在对应关系。批监听器仍然支持异步返回类型。