通过同一个 ActiveMQConnectionFactory 使用多个 ActiveMQ 队列答案

【问题标题】：Consume multiple ActiveMQ queues via the same ActiveMQConnectionFactory通过同一个 ActiveMQConnectionFactory 使用多个 ActiveMQ 队列
【发布时间】：2014-09-09 10:59:06
【问题描述】：

我的应用程序处理放置在单个 ActiveMQ 队列（名为“incoming.queue”）中的传入消息。我有一个处理消息的 MessageListener，一切运行良好。我的 Java 配置如下：

@Configuration
@ComponentScan(basePackages="uk.co.domain")
public class JmsConfig {

    @Bean
    public ActiveMQConnectionFactory connectionFactory() {
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory();
        activeMQConnectionFactory.setBrokerURL("tcp://localhost:61616");
        return activeMQConnectionFactory;
    }

    @Bean
    public DefaultMessageListenerContainer jmsListenerContainer() {
        DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
        dmlc.setConnectionFactory(connectionFactory());
        dmlc.setDestination(new ActiveMQQueue("incoming.queue"));
        dmlc.setMessageListener(new QueueProcessor());
        dmlc.setConcurrentConsumers(50);
        return dmlc;
    }
}



public class QueueProcessor implements MessageListener {

    public void onMessage(Message message) {

        // process the message
    }
}

另一个部门正在对上游进行更改，以便消息将在三个不同的队列中传播，名为“high_priority.queue”、“med_priority.queue”和“low_priority.queue”。每个队列的并发消费者数量需要分别为 50、20 和 5 个。

在我的代码中，相同的QueueProcessor 将负责处理消息，但我不确定如何修改我的配置以创建三个消息侦听器而不是一个。任何建议表示赞赏。

【问题讨论】：

保持解耦的正确做法是让上游人员使用优先级队列来使用内置的 JMS 优先级，然后您就不必进行任何更改。没有理由根据优先级拆分队列。
我同意你的评论，但不幸的是我无法控制上游的变化。

标签： java spring jms activemq

【解决方案1】：

我通过简单地创建多个 bean 解决了这个问题 - 没想到会这么容易：

@Bean
public DefaultMessageListenerContainer highPriorityQueue() {
    DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
    dmlc.setConnectionFactory(connectionFactory());
    dmlc.setDestination(new ActiveMQQueue("high_priority.queue"));
    dmlc.setMessageListener(new QueueProcessor());
    dmlc.setConcurrentConsumers(50);
    return dmlc;
}

@Bean
public DefaultMessageListenerContainer medPriorityQueue() {
    DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
    dmlc.setConnectionFactory(connectionFactory());
    dmlc.setDestination(new ActiveMQQueue("med_priority.queue"));
    dmlc.setMessageListener(new QueueProcessor());
    dmlc.setConcurrentConsumers(20);
    return dmlc;
}

@Bean
public DefaultMessageListenerContainer lowPriorityQueue() {
    DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
    dmlc.setConnectionFactory(connectionFactory());
    dmlc.setDestination(new ActiveMQQueue("low_priority.queue"));
    dmlc.setMessageListener(new QueueProcessor());
    dmlc.setConcurrentConsumers(5);
    return dmlc;
}

【讨论】：

【解决方案2】：

您可以创建一个通用处理器类，并使用构造函数在该类上设置您的连接、目的地和其他配置，方法是将队列名称和消费者计数作为实现此通用类的处理器类的输入。你也可以将这个类的构造函数声明为spring conf的bean。

【讨论】：