Spring Boot+IBM MQ多JMS Listener实例串行处理异常排查

我正在基于Spring Boot、JMS Listener和IBM MQ构建消息应用，需同时运行同一监听器的两个实例，要求第二个实例需等待第一个实例完全处理并确认消息后再执行。

配置类代码 #

@Component
@EnableJMS
public class jmsconfig{

 @Bean
 public MQConnectionfactory getConnectionFactory(){
     MQConnectionFactory connectionFactory = new MQConnectionFactory();
     connectionFactory.setQueueManagerName("AA");
     connectionFactory.setExpirationTimeOut(3600000);
     connectionFactory.setPerformExpiration(true);
     connectionFactory.setPerformValidation(true);
     connectionFactory.setPerformOptimalSizeCheck(false);
     connectionFactory.setValidationTimeOut(180000);
     connectionFactory.setMinIdle(3);
     connectionFactory.setMaxIdle(5);
     reurn connectionFactory ;
}

 @Bean(name="jmsListenerContainerFactory")
 public JmsListenerContainerFactory jmsListenerContainerFactory(MQConnectionfactory mqConnectionfactory){
     DefaultJmsListenerContainerFactory  containerFactory = new DefaultJmsListenerContainerFactory();
     containerFactory.setConnectionFactor(mqConnectionfactory);
     containerFactory.setSessionTransacted(true);
     containerFactory.setSessionAcknowledgementMode(Session.CLIENT_ACKNOWLEDE);
     reurn containerFactory ;
}

}

监听器代码 #

@Component
public class Receiver {
      
  public sttaic int count = 0;

  @JmsListener(destination = "${inwardQueueName}", containerFactory = "jmsListenerContainerFactory")
  public void receiveMessage(javax.jms.Message message) throws javax.jms.JMSException {
    String messagetxt = "";
    OrderObject order = null;
    if (message instanceof javax.jms.TextMessage) {
      messagetxt = ((TextMessage)message).getText();
      OrderObject order = //code to covert messagetxt to object
      System.out.println("Message pciked up with Order Id : " +order.getOrderId);
      TimeUnit.SECONDS.sleep(15);
      count++;
     if(order.getOrderId==1){
      if(count<=4){
        throw new Exception("Exception occurred");
      }
    }
    }
    TimeUnit.SECONDS.sleep(15);
    System.out.println("Message Acknowledged for orderId: " +order.getOrderId);
    message.acknowledge();
  }
}

测试过程 #

• 启动第一个实例并发送orderId为1的消息，控制台打印“Message picked up with Order Id : 1”后等待15秒；
• 随即启动第二个实例并发送orderId为2的消息，第二个实例立即拾取并处理该消息，控制台输出“Message picked up with orderId 2”和“Message Acknowledged for orderId: 2”；
• 待第一个实例处理完成后，控制台多次打印orderId为1的拾取信息并最终确认。

当前问题 #

第二个实例会在第一个实例处理消息时并行拾取并处理消息，不符合预期，请求排查该实现问题。

问题原因分析 #

1. 队列默认分发机制：IBM MQ队列默认采用负载均衡策略，会将消息分发给所有在线的消费者实例，只要队列有可用消息，每个实例都会尝试获取，这是JMS队列的标准行为。
2. 配置冲突：同时设置setSessionTransacted(true)和Session.CLIENT_ACKNOWLEDE是互斥的——事务模式下，消息的确认由事务提交/回滚控制，手动调用message.acknowledge()完全无效，还会导致配置逻辑混乱。
3. 代码拼写与语法错误：配置类和监听器存在多处拼写错误（如reurn、setConnectionFactor、CLIENT_ACKNOWLEDE、sttaic等），以及语法问题（重复定义变量、抛出非声明异常等），这些会导致配置不生效或运行异常。

解决方案 #

要实现多实例下同一时间仅一个实例处理消息，需从MQ队列配置和Spring JMS配置两方面调整：

1. 设置IBM MQ队列独占访问

在MQ控制台找到目标队列，将Exclusive属性设为Yes。该配置会让队列同一时间仅允许一个消费者连接获取消息，其他实例会处于等待状态，直到当前消费者释放独占锁。

2. 修正Spring JMS容器配置

移除冲突的确认模式，统一使用事务模式，并限制单实例消费者数量：

@Component
@EnableJMS
public class JmsConfig {

    @Bean
    public MQConnectionFactory getConnectionFactory(){
        MQConnectionFactory connectionFactory = new MQConnectionFactory();
        connectionFactory.setQueueManagerName("AA");
        connectionFactory.setExpirationTimeOut(3600000);
        connectionFactory.setPerformExpiration(true);
        connectionFactory.setPerformValidation(true);
        connectionFactory.setPerformOptimalSizeCheck(false);
        connectionFactory.setValidationTimeOut(180000);
        connectionFactory.setMinIdle(3);
        connectionFactory.setMaxIdle(5);
        return connectionFactory ;
    }

    @Bean(name="jmsListenerContainerFactory")
    public JmsListenerContainerFactory<?> jmsListenerContainerFactory(MQConnectionFactory mqConnectionFactory){
        DefaultJmsListenerContainerFactory containerFactory = new DefaultJmsListenerContainerFactory();
        containerFactory.setConnectionFactory(mqConnectionFactory);
        containerFactory.setSessionTransacted(true);
        // 事务模式下无需手动确认，移除CLIENT_ACKNOWLEDGE配置
        containerFactory.setConcurrency("1"); // 限制单实例仅启动1个消费者，避免单实例内并行处理
        return containerFactory ;
    }
}

3. 修正监听器代码错误

@Component
public class Receiver {

    public static int count = 0;

    @JmsListener(destination = "${inwardQueueName}", containerFactory = "jmsListenerContainerFactory")
    public void receiveMessage(javax.jms.Message message) throws javax.jms.JMSException {
        String messagetxt = "";
        OrderObject order = null;
        if (message instanceof javax.jms.TextMessage) {
            messagetxt = ((javax.jms.TextMessage)message).getText();
            order = // code to convert messagetxt to object
            System.out.println("Message picked up with Order Id : " + order.getOrderId());
            try {
                TimeUnit.SECONDS.sleep(15);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new JMSException("Processing interrupted");
            }
            count++;
            if(order.getOrderId() == 1){
                if(count <=4){
                    throw new JMSException("Exception occurred");
                }
            }
        }
        try {
            TimeUnit.SECONDS.sleep(15);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new JMSException("Processing interrupted");
        }
        System.out.println("Message Acknowledged for orderId: " + order.getOrderId());
        // 事务模式下无需手动调用acknowledge，事务提交后自动确认
    }
}

补充说明

如果不需要全局独占队列，也可以使用消息分组：为需要串行处理的消息设置相同的Group ID，MQ会确保同一分组的消息仅由一个消费者处理，适用于特定消息序列的串行需求。

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