Camel同JMS队列收发消息循环消费问题：如何控制仅消费一次？

问题描述 #

我们的系统已配置为从同一队列消费并发送回复，即JMSDestination与JMSReplyTo相同，且目前无法修改该配置。

在集成测试中遇到了两难情况：

• 若设置replyToSameDestinationAllowed=true，Camel会持续消费发送到队列的回复，陷入循环无法停止；
• 若不设置该参数，Camel会拒绝发送回复，并提示：

JMSDestination and JMSReplyTo is the same, will skip sending a reply message to itself

我希望在独立方法中消费消息并进行断言，如何让Camel仅消费一次消息、忽略后续消息？

我的路由末尾调用了stop()以自动发送回复。接收第二条消息（即回复）时，日志显示：

2023-01-10 14:37:22,186 DEBUG [org.apa.cam.com.jms.EndpointMessageListener]-{Camel (camel-1) thread #19 - JmsConsumer[my.queue]}-Received Message has JMSCorrelationID [ID:hostname-1673354133272-4:1:1:10:1]

能否利用JMSCorrelationID忽略回复？或是应该停止路由、回滚？该如何处理？

解决方案 #

1. 用JMSCorrelationID做消息过滤

直接在JMS消费者端点添加消息选择器，只消费没有关联ID、或关联ID与初始请求不匹配的消息，自动过滤路由发送的回复。

配置示例：

from("jms:queue:my.queue?messageSelector=JMSCorrelationID IS NULL OR JMSCorrelationID != '初始请求的CorrelationID'")
    .process(yourRequestProcessor)
    .stop();

测试时可以先记录初始请求的JMSCorrelationID，再动态设置选择器；如果用Camel测试框架，也能临时修改端点的选择器配置。

2. 消费一次后停止路由

在测试方法中，完成第一条消息的断言后立刻停止对应路由，阻止后续消费：

// 发送测试请求
template.sendBody("jms:queue:my.queue", "test-request");

// 消费并断言初始消息
Exchange exchange = consumer.receive("jms:queue:my.queue", 5000);
assertNotNull(exchange);
assertEquals("test-request", exchange.getIn().getBody());

// 停止路由，终止循环
context.stopRoute("your-target-route-id");

测试结束后可以调用context.startRoute("your-target-route-id")恢复路由，清理测试环境。

3. 用标记跳过重复消费

在路由中加入线程安全的标记，记录是否已处理初始请求，检测到回复消息时直接跳过：

// 测试类中定义线程安全标记
private final AtomicBoolean hasProcessedInitial = new AtomicBoolean(false);

// 路由配置
from("jms:queue:my.queue?replyToSameDestinationAllowed=true")
    .process(exchange -> {
        String correlationId = exchange.getIn().getHeader(JMSCorrelationID, String.class);
        // 是回复消息且已处理过初始请求，直接终止路由
        if (correlationId != null && hasProcessedInitial.get()) {
            exchange.setProperty(Exchange.ROUTE_STOP, true);
            return;
        }
        // 处理初始请求逻辑
        processInitialRequest(exchange);
        // 标记已处理
        hasProcessedInitial.set(true);
    })
    .stop();

4. 测试环境临时修改JMSReplyTo

如果测试场景允许控制请求发送逻辑，可以把测试请求的JMSReplyTo设为临时队列，让回复不回原队列：

Message testMsg = template.getDefaultMessage();
testMsg.setBody("test-request");
testMsg.setHeader(JMSReplyTo, "jms:queue:test-temp-queue");
template.send("jms:queue:my.queue", testMsg);

// 从临时队列接收回复并断言
Exchange replyExchange = consumer.receive("jms:queue:test-temp-queue", 5000);

推荐选择 #

如果初始请求没有设置JMSCorrelationID，方案1的消息选择器是最简洁的；如果需要快速终止循环，方案2的停止路由在测试场景下直观可控。

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