如何使用自动连线的Spring Boot监听多个队列？

Z时代
2024-01-10
分类：问答

我是Spring Boot" title="Spring Boot">Spring Boot的新手，并且正在玩它。目前，我已经构建了一些应用，希望能够通过队列相互通信。我目前有一个侦听器对象，可以从特定队列接收消息。

@Configuration
public class Listener {
    final static String queueName = "myqueue";
    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(queueName);
        container.setMessageListener(listenerAdapter);
        return container;
    }
    @Bean
    Receiver receiver() {
        return new Receiver();
    }
    @Bean
    MessageListenerAdapter listenerAdapter(Receiver receiver) {
        return new MessageListenerAdapter(receiver, "receiveMessage");
    }
}

这可行。但是，现在我希望能够收听另一个队列。所以我想我会复制上面的对象并更改队列名称。不幸的是，这没有用，因为Spring

boot只为其中一个创建连接。关于如何让Spring Boot应用程序侦听多个队列的任何想法？

回答：

好的，我想出了如何让它侦听多个队列。认为与我的其他解决方案相比，可能存在一些缺点，主要是如果列出的队列不存在，则它不起作用。我最终使用了@RabbitListener完全不同的方法

@Component
public class EventListener {
    private static Logger LOG = LoggerFactory.getLogger(EventListener.class);
    private CountDownLatch latch = new CountDownLatch(1);
    @RabbitListener(queues = "myqueue")
    public void processPaymentMessage(Object message) {
        LOG.info("Message is of type: " + message.getClass().getName());
        if(!(message instanceof byte[])) message = ((Message) message).getBody();
        String content = new String((byte[])message, StandardCharsets.UTF_8);
        LOG.info("Received on myqueue: " + content);
        latch.countDown();
    }
    @RabbitListener(queues = "myotherqueue")
    public void processOrderMessage(Object message) {
        LOG.info("Message is of type: " + message.getClass().getName());
        if(!(message instanceof byte[])) message = ((Message) message).getBody();
        String content = new String((byte[])message, StandardCharsets.UTF_8);           
        LOG.info("Received on myotherqueue: " + content);
        latch.countDown();
    }   
}