大量带有选择器 JMSCorrelationID='CamelDummyJmsMessageSelector' 的异步消息使用者

Question

我正在使用 apache Camel 通过 spring-jms 和 ibm-mq 配置 jms 请求/回复模式。 我的 Camel Route 配置为 InOut 模式，在将消息发送到请求队列后，我能够成功从回复队列读取消息。 然而，在成功处理消息后，有一个后台骆驼线程正在运行，它正在为已处理的同一回复队列创建消息消费者。以下日志消息正在被写入无限。 收到回复后，我无法理解为什么 其他线程仍在尝试使用该消息。就我而言，我只会从回复队列中收到 1 条消息。

我们的 MQ 管理员还提到，队列上创建了许多异步消费者，并且处于 SUSPENDED 状态。 https://www.ibm.com/docs/en/ibm-mq/9.3?topic=reference-display-qstatus-display-queue-status

在调试时，我了解到代码正在进入以下方法，但不确定进程将在哪种情况下终止。

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（4.7.0）和spring-boot（3.2.6）库

Answer 1

您的 jms 中似乎存在一些配置问题，更准确地说，要使用的并发消费者的数量，我会强制只有一个。

类似这样的事情：

import org.apache.camel.component.jms.JmsConfiguration;
import org.apache.camel.component.jms.JmsEndpoint;
import org.apache.camel.component.jms.reply.ReplyManager;
import org.apache.camel.component.jms.reply.SharedQueueMessageListenerContainer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class CamelConfig {

    @Bean
    public JmsConfiguration jmsConfiguration(CachingConnectionFactory cachingConnectionFactory) {
        JmsConfiguration jmsConfiguration = new JmsConfiguration();
        jmsConfiguration.setConnectionFactory(cachingConnectionFactory);
        jmsConfiguration.setReplyToMaxConcurrentConsumers(1); // Ensure only one consumer
        return jmsConfiguration;
    }

    @Bean
    public JmsEndpoint jmsEndpoint(JmsConfiguration jmsConfiguration) {
        JmsEndpoint jmsEndpoint = new JmsEndpoint();
        jmsEndpoint.setConfiguration(jmsConfiguration);
        return jmsEndpoint;
    }

    @Bean
    public ReplyManager replyManager(JmsEndpoint jmsEndpoint) {
        SharedQueueMessageListenerContainer listenerContainer = new SharedQueueMessageListenerContainer();
        listenerContainer.setConcurrentConsumers(1); // Ensure only one consumer
        listenerContainer.setIdleConsumerLimit(1); // Limit idle consumers
        listenerContainer.setIdleTaskExecutionLimit(1); // Limit idle task executions
        return new ReplyManager(jmsEndpoint, listenerContainer);
    }
}