Spring JMS并发消费配置疑问：已设并发数但仅单实例消费

看起来你遇到的问题很典型——明明配置了5个并发消费者，结果只有一个在干活，剩下的全闲着。结合你用编程式注册队列的场景，我来帮你排查几个关键问题：

1. 最可能的元凶：队列预取策略“抢”走了所有消息 #

绝大多数JMS Broker（比如ActiveMQ）默认会给消费者设置一个超大的预取数（比如ActiveMQ默认是1000）。当第一个消费者启动后，它会一次性把队列里的1000多条消息全预取到本地缓存，剩下的消费者自然没消息可处理，只能原地等待。

解决方案：
调整预取策略，限制每个消费者的预取数量，让消息能均匀分配给多个消费者：

• 如果用ActiveMQ，可以直接在连接工厂URL里配置：

  tcp://your-broker-url:61616?jms.prefetchPolicy.queuePrefetch=10
  

• 或者通过代码设置：

  ActiveMQConnectionFactory connectionFactory = (ActiveMQConnectionFactory) defaultJMSListenerConnectionFactory;
  PrefetchPolicy prefetchPolicy = new PrefetchPolicy();
  prefetchPolicy.setQueuePrefetch(10); // 数值可以根据业务场景调整
  connectionFactory.setPrefetchPolicy(prefetchPolicy);
  

2. 检查并发配置是否真正生效 #

你提到配置了5个并发消费者，但要确认这个配置是否传递到了每个注册的Endpoint对应的容器里。从你的代码看，你直接用defaultJMSListenerConnectionFactory注册Endpoint，但并发数是在容器工厂（DefaultJmsListenerContainerFactory）里配置的，不是直接在连接工厂上设置。

修正后的代码示例：

// 先创建并配置容器工厂
DefaultJmsListenerContainerFactory containerFactory = new DefaultJmsListenerContainerFactory();
containerFactory.setConnectionFactory(defaultJMSListenerConnectionFactory);
containerFactory.setConcurrency("5-10"); // 这里设置你的并发范围
containerFactory.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE); // 按需调整消息确认模式
// 如果不需要事务，直接关闭避免消息锁定
containerFactory.setSessionTransacted(false);

// 用配置好的工厂注册所有Endpoint
int counter = 0;
queues.forEach(queue -> {
    SimpleJmsListenerEndpoint endpoint = new SimpleJmsListenerEndpoint();
    endpoint.setId("myJmsEndpoint" + counter++);
    endpoint.setDestination(queue);
    // 别忘了绑定你的消息处理器
    endpoint.setMessageListener(message -> {
        // 你的消息消费逻辑
    });
    registrar.registerEndpoint(endpoint, containerFactory);
});

3. 事务配置导致消息锁定 #

如果你的消费者开启了事务（默认可能是开启状态），但消息处理过程中事务没有及时提交，那么第一个消费者获取的消息会一直处于锁定状态，其他消费者根本拿不到。

解决方案：

• 如果业务不需要事务，直接关闭容器的事务：containerFactory.setSessionTransacted(false);
• 如果必须用事务，确保你的消息处理逻辑能正常提交事务，避免长时间持有消息。

4. 其他排查小细节 #

• 确认你的队列是点对点队列（不是主题），主题是广播模式，每个消费者都会收到消息，不会出现这种“抢消息”的情况；
• 检查Broker的配置，有没有给队列设置独占消费者属性，这会强制只有一个消费者能连接到队列。

按照上面的步骤排查，应该能让你的5个DMLC消费者同时跑起来！

内容的提问来源于stack exchange，提问作者Sach