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 s。我还在 Spring 容器启动中验证了只有一个(预期的) rabbitListenerContainerFactory bean 。

Answer 1

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

你的听众长什么样？

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

您要么需要添加containerFactory @RabbitListener 的属性(property)或者您需要将您的 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