Camel+Spring-JMS+IBM MQ InOut模式出现消费者残留问题

问题背景 #

使用Apache Camel 4.7.0 + Spring Boot 3.2.6，结合Spring-JMS与IBM MQ配置JMS请求/响应模式，Camel路由采用InOut模式。成功发送消息到请求队列并读取回复队列的消息后，后台Camel线程仍持续为已处理的回复队列创建消息消费者，无限输出TRACE级日志。MQ管理员反馈队列上存在大量处于SUSPENDED状态的异步消费者，调试发现代码进入org.springframework.jms.listener.DefaultMessageListenerContainer.AsyncMessageListenerInvoker#executeOngoingLoop方法，但流程未自动终止。

日志信息 #

26-08-2024 13:42:55.270 [Camel (camel-1) thread #6 - QueueReplyManager["TEST"]] TRACE org.apache.camel.component.jms.reply.SharedQueueMessageListenerContainer: Consumer [Cached JMS MessageConsumer: com.ibm.mq.jakarta.jms.MQQueueReceiver@3a9df93c] of session [Cached JMS Session: {"ConnectionId":"C3E2D8C3D4D8E4F14040404040404040DF9A9ECAB3810001","ObjectId":"com.ibm.msg.client.jakarta.wmq.internal.WMQSession@593ea8be","Channel":"TEST","Port":"1431","ConnectionMode":"WMQ_CM_CLIENT","Host":"CHUBBHOST1.chubb.com/10.64.25.1","ResolvedQueueManager":"MQU1","QueueManager":"MQU1"}] did not receive a message
26-08-2024 13:42:55.270 [Camel (camel-1) thread #6 - QueueReplyManager["TEST"]] TRACE org.apache.camel.component.jms.reply.SharedQueueMessageListenerContainer: runningAllowed() -> true
26-08-2024 13:42:55.270 [Camel (camel-1) thread #6 - QueueReplyManager["TEST"]] TRACE org.apache.camel.component.jms.reply.JmsReplyHelper: Using MessageSelector[JMSCorrelationID='CamelDummyJmsMessageSelector']
26-08-2024 13:42:55.862 [Camel (camel-1) thread #5 - JmsReplyManagerTimeoutChecker["TEST"]] TRACE org.apache.camel.component.jms.reply.CorrelationTimeoutMap: Running purge task to see if any entries have been timed out
26-08-2024 13:42:55.862 [Camel (camel-1) thread #5 - JmsReplyManagerTimeoutChecker["TEST"]] TRACE org.apache.camel.component.jms.reply.CorrelationTimeoutMap: There are 0 in the timeout map

核心原因 #

Camel JMS组件在InOut模式下默认启用共享回复管理器（SharedQueueReplyManager），该管理器会持续维护消费者监听回复队列，即使没有待处理请求。同时Spring JMS的DefaultMessageListenerContainer异步调用循环会持续运行，直到容器被显式停止或触发配置的终止条件。

解决办法 #

1. 配置回复管理器空闲超时 #

通过replyManagerIdleTimeout参数，让回复管理器在空闲指定时间后自动关闭消费者、释放资源：

@Bean
public JmsComponent jmsComponent(ConnectionFactory connectionFactory) {
    JmsComponent component = new JmsComponent();
    component.setConnectionFactory(connectionFactory);
    // 设置30秒空闲超时
    component.setReplyManagerIdleTimeout(30000);
    return component;
}

或通过配置文件：

camel:
  component:
    jms:
      reply-manager-idle-timeout: 30000

2. 禁用共享回复管理器 #

配置useSharedReplyManager=false，让每个请求创建独立的回复消费者，请求完成后自动销毁（适合请求量不大的场景，会增加消费者创建/销毁开销）：

component.setUseSharedReplyManager(false);

或配置文件：

camel:
  component:
    jms:
      use-shared-reply-manager: false

3. 自定义Spring JMS容器配置 #

针对DefaultMessageListenerContainer设置接收超时和停止超时，控制消费者循环终止逻辑：

@Bean
public DefaultMessageListenerContainer replyMessageListenerContainer(ConnectionFactory connectionFactory) {
    DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);
    container.setDestinationName("TEST.REPLY.QUEUE");
    // 设置接收超时（5秒），超时后检查是否继续运行
    container.setReceiveTimeout(5000);
    // 设置停止超时（10秒），确保容器及时关闭
    container.setStopTimeout(10000);
    return container;
}

若为Camel自动创建的容器，可通过JMS组件的listenerContainerFactory注入自定义配置。

4. 临时调整日志级别 #

若暂时无法彻底解决，先降低相关包的日志级别避免刷屏：

logging:
  level:
    org.apache.camel.component.jms.reply: INFO

验证标准 #

修改配置后需确认：

• 无消息接收的TRACE日志不再持续输出
• MQ队列的SUSPENDED状态消费者数量不再持续增长
• 空闲一段时间后，回复队列的消费者被自动销毁

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