這是我的@Configuration @Bean public AmqpAdmin amqpAdmin() { RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory()); DirectExchange dirExchange = new DirectExchange("evtExchange", true, false); rabbitAdmin.declareExchange(dirExchange); rabbitAdmin.declareQueue(processQueue); Binding processBinding = BindingBuilder.bind(processQueue) .to(dirExchange).with("rkey.process"); rabbitAdmin.declareBinding(processBinding); return rabbitAdmin; } @Bean public RabbitTemplate rabbitTemplate() { RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory()); return rabbitTemplate; } @Bean public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() { SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); factory.setConnectionFactory(connectionFactory()); SimpleMessageListenerContainer container = factory .createListenerContainer(); factory.setConcurrentConsumers(50); factory.setMaxConcurrentConsumers(100); container.setStartConsumerMinInterval(3000); container.setQueues(processQueue); factory.setAdviceChain(retryInterceptor()); return factory; } @Bean public RetryOperationsInterceptor retryInterceptor() { return RetryInterceptorBuilder.stateless().maxAttempts(5) .backOffOptions(1000, 2.0, 10000).recoverer(new RejectAndDontRequeueRecoverer()).build(); } @Bean public ProcessQueueListener processListener() { return new ProcessQueueListener(); }只有當(dāng)我單獨(dú)實(shí)例化時(shí)processListener()，processListener2()我才processListener3()開始在 RabbitMQ Admin 中看到進(jìn)程隊(duì)列的多個(gè)消費(fèi)者，并且每個(gè)偵聽器都在處理消息，否則我只看到一個(gè)消費(fèi)者，盡管指定了setConcurrentConsumers()有沒有一種優(yōu)雅的方式來按需聲明多個(gè)監(jiān)聽器，根據(jù)需要增加和減少。或者聲明多個(gè)@Beans 是唯一的選擇？還是我做錯(cuò)了什么？
1 回答
達(dá)令說
TA貢獻(xiàn)1821條經(jīng)驗(yàn) 獲得超6個(gè)贊
你用的是什么版本？
我剛剛復(fù)制了你的容器工廠，它對(duì)我來說很好（2.1.3）......
順便說一句，從 2.0 版開始，您可以添加concurrency到@RabbitListener，它將覆蓋容器工廠中的任何值。
/**
* Set the concurrency of the listener container for this listener. Overrides the
* default set by the listener container factory. Maps to the concurrency setting of
* the container type.
* <p>For a
* {@link org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer
* SimpleMessageListenerContainer} if this value is a simple integer, it sets a fixed
* number of consumers in the {@code concurrentConsumers} property. If it is a string
* with the form {@code "m-n"}, the {@code concurrentConsumers} is set to {@code m}
* and the {@code maxConcurrentConsumers} is set to {@code n}.
* <p>For a
* {@link org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer
* DirectMessageListenerContainer} it sets the {@code consumersPerQueue} property.
* @return the concurrency.
* @since 2.0
*/
String concurrency() default "";
此外，不相關(guān)，但您不應(yīng)該rabbitAdmin.declareExchange(dirExchange)在 bean 聲明中這樣做 - 在應(yīng)用程序上下文生命周期中連接到 RabbitMQ 還為時(shí)過早。將交換、隊(duì)列和綁定添加為@Beans，管理員會(huì)自動(dòng)找到并聲明它們。
添加回答
舉報(bào)
0/150
提交
取消