java - Spring AMQP 在队列 Bean 上设置 prefetchCount

Question

如何为此队列使用者设置 prefetchCount？

@Bean
public Queue eventQueue(AmqpAdmin amqpAdmin) {
    Queue queue = QueueBuilder.durable(EVENT_QUEUE_NAME)
            ...
            .build();

    TopicExchange topicExchange = new TopicExchange(TOPIC_EXCHANGE, true, false);

    amqpAdmin.declareBinding(BindingBuilder
            .bind(queue)
            .to(topicExchange)
            .with(EVENT_ROUTING_KEY));

    return queue;
}

documentation注意到 prefetchCount 这是一个容器配置，但在我的工厂 bean 上设置它不起作用，并且该值默认为 250:

    @Bean
    public SimpleRabbitListenerContainerFactory containerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setPrefetchCount(10); // doesn't work; defaults to 250
        return factory;
    }

更新

根据下面@GaryRussell 的评论，我测试了默认的 rabbitListenerContainerFactory 并验证了我的 Spring Boot 配置 spring.rabbitmq.listener.simple.prefetch 正在被消耗AbstractRabbitListenerContainerFactoryConfigurer。但是，当我查看 RabbitMQ 中的队列使用者时，我可以看到使用默认容器设置定义的队列的 prefetchCount 仍然为 250:

我使用 RabbitMQ 管理面板作为事实来源。我不认为这是在撒谎，因为我有一堆用自定义容器实例化的动态队列，并且它们确实具有非默认(正确)的 prefetchCount 。我还在 Spring 容器启动中验证了只有一个(预期的)rabbitListenerContainerFactory bean。

Answer 1

预取不是队列属性，而是消费者属性。

你的听众是什么样的？

您正在为容器工厂使用非标准名称。

您需要将 containerFactory 属性添加到 @RabbitListener，或者需要将您的 bean 重命名为 rabbitListenerContainerFactory (覆盖 Boot 定义的工厂 @Bean)。

还有

amqpAdmin.declareBinding(BindingBuilder

您不应该在 bean 定义中与代理交谈 - 现在还为时过早。

只需将队列、交换和绑定(bind)添加为 @Bean 并 您还可以在 application.properties/yaml 文件中设置预取(如果您使用的是 Spring Boot)。管理员将找到它们并在连接首次打开时声明它们。

编辑

还有其他事情发生...

@SpringBootApplication
public class So62049769Application {

    public static void main(String[] args) {
        SpringApplication.run(So62049769Application.class, args);
    }

    @Bean
    public Queue queue() {
        return new Queue("so62049769");
    }

    @RabbitListener(queues = "so62049769")
    public void listen(String in) {
        System.out.println(in);
    }

}

spring.rabbitmq.listener.simple.prefetch=42