java - JMS 发布者/订阅者模型中的并发消费者

Question

我正在使用具有 DefaultMessageListenerContainer 的 pub/sub 模型。我已将并发消费者配置为 5 个。如何唯一标识每个消费者？

我试图将相应监听器处理的每个事件存储在 map 中。这是我努力寻找哪个消费者处理了我的事件的部分。我怎样才能做到这一点？

关键是如何在并发消费者的情况下唯一标识一个消费者。我的 DMLC 配置是

@Bean
    public DefaultMessageListenerContainer listenerContainers() {
        DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
        container.setConnectionFactory(connectionFactory());
        container.setDestinationName(COMMENT_QUEUE);
        container.setPubSubDomain(true);
        container.setSessionTransacted(true);
        container.setConcurrentConsumers(5);
        container.setSubscriptionDurable(true);
        container.setMessageListener(datafileSubscriber);
        container.start();
        return container;
    } 

如您所知，这将使 datafilesubscriber 在 5 个不同的线程中执行。编辑 :我听说我们需要一个具有单独客户端 ID 的连接工厂。

  @Bean
        public ActiveMQConnectionFactory connectionFactory(){
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
            connectionFactory.setBrokerURL(DEFAULT_BROKER_URL);
            connectionFactory.setClientID("subscriber");
            return connectionFactory;
        }

所以现在我的问题是我应该如何处理 dmlc 中的 connectionFactory 参数？我应该使用相同的connectionFactory bean 还是创建5 个具有不同客户端ID 的不同的connectionFactory bean？

编辑2:我的示例代码有两个订阅同一主题的监听器，并且都是持久消费者。当消息发布到主题时，两者都会收到该消息，其中一个会处理该消息，而另一个会忽略该消息，因为它在缓存中看到另一个监听器处理了该消息。我的配置是

@Bean
    public ActiveMQConnectionFactory connectionFactory(){
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        connectionFactory.setBrokerURL(DEFAULT_BROKER_URL);
        return connectionFactory;
    }
@Bean
    public DefaultMessageListenerContainer listenerContainers() {
        DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
        container.setConnectionFactory(connectionFactory());
        //container.setConnectionFactory(connectionFactory1());
        container.setClientId("consumer1");
        container.setDestinationName(COMMENT_QUEUE);
        container.setPubSubDomain(true);
        container.setSessionTransacted(true);
        container.setSubscriptionDurable(true);
        container.setMessageListener(datafileSubscriber);
        container.start();
        return container;
    }

    @Bean
    public DefaultMessageListenerContainer listenerContainers1() {
        DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
        container.setConnectionFactory(connectionFactory());
        container.setClientId("consumer2");
        container.setDestinationName(COMMENT_QUEUE);
        container.setPubSubDomain(true);
        container.setSessionTransacted(true);
        container.setSubscriptionDurable(true);
        container.setMessageListener(datafileSubscriber);
        container.start();
        return container;
    } 

我的问题仍然相同，由于两个监听器执行相同的操作，因此如何确定正在处理哪个监听器？

还告诉我我的配置是否正确？

Answer 1

对主题使用并发没有多大意义，因为消息监听器将收到相同的消息 5 次。

如果这确实是您想要的，您需要 5 个并发=1 的容器。

标准 JMS 不支持主题上的竞争消费者(某些代理对此有扩展，但您必须参阅代理的文档)。