【发布时间】：2019-08-02 14:55:49
【问题描述】：
我在 InOut 模式中有多个顺序队列，每个队列通向一个 Camel 处理器。 如果处理器花费太多时间进行处理，我希望请求-回复检测到超时并发送 ExchangeTimedOutException。 CamelJmsRequestTimeout 的超时有效，但仅适用于第一个队列。
例如， 如果我有这个：
q1 -> p1 -> q2 -> p2 -> q3 -> p3
q1 CamelJmsRequestTimeout 例如是 5s。如果每个处理器需要 3 秒（我用 sleep() 模拟处理器的延迟），我将在 p2 超时，因为总时间是 6 秒。为什么对 q1 的回复没有在 p1 之后发送，而是在 p3 之后发送？
这是一个示例代码：
from("jms:queue:q1")
.setExchangePattern(ExchangePattern.InOut)
.setHeader(JmsConstants.JMS_REQUEST_TIMEOUT, constant("5000"))
.process("p1")
.to("jms:queue:q2");
from("jms:queue:q2")
.setExchangePattern(ExchangePattern.InOut)
.setHeader(JmsConstants.JMS_REQUEST_TIMEOUT, constant("5000"))
.process("p2")
.to("jms:queue:q3");
from("jms:queue:q3")
.setExchangePattern(ExchangePattern.InOut)
.setHeader(JmsConstants.JMS_REQUEST_TIMEOUT, constant("5000"))
.process("p3");
我的设计可行吗？如果是，我应该使用哪个选项？
谢谢
编辑：
2019-08-07 09:45:02,718 [main] INFO SpringCamelContext - Apache Camel 2.21.0 (CamelContext: camel-1) started in 1.152 seconds
2019-08-07 09:45:05,973 [Camel (camel-1) thread #1 - JmsConsumer[q1]] INFO input - Exchange[ExchangePattern: InOut, BodyType: String, Body: stuff in the body]
2019-08-07 09:45:09,022 [Camel (camel-1) thread #2 - JmsConsumer[q2]] INFO input - Exchange[ExchangePattern: InOut, BodyType: String, Body: stuff in the body]
2019-08-07 09:45:11,991 [Camel (camel-1) thread #25 - JmsReplyManagerOnTimeout[q2]] WARN TemporaryQueueReplyManager - Timeout occurred after 5000 millis waiting for reply message with correlationID [Camel-ID-ITEM-S69138-1565163898705-0-2] on destination temp-queue://ID:ITEM-S69138-55901-1565163899733-4:20:1. Setting ExchangeTimedOutException on (MessageId: ID:ITEM-S69138-55901-1565163899733-4:19:1:1:1 on ExchangeId: ID-ITEM-S69138-1565163898705-0-1) and continue routing.
2019-08-07 09:45:11,996 [Camel (camel-1) thread #1 - JmsConsumer[q1]] WARN EndpointMessageListener - Execution of JMS message listener failed. Caused by: [org.apache.camel.RuntimeCamelException - org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 5000 millis due reply message with correlationID: Camel-ID-ITEM-S69138-1565163898705-0-2 not received on destination: temp-queue://ID:ITEM-S69138-55901-1565163899733-4:20:1. Exchange[ID-ITEM-S69138-1565163898705-0-1]]
org.apache.camel.RuntimeCamelException: org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 5000 millis due reply message with correlationID: Camel-ID-ITEM-S69138-1565163898705-0-2 not received on destination: temp-queue://ID:ITEM-S69138-55901-1565163899733-4:20:1. Exchange[ID-ITEM-S69138-1565163898705-0-1]
at org.apache.camel.util.ObjectHelper.wrapRuntimeCamelException(ObjectHelper.java:1830)
at org.apache.camel.component.jms.EndpointMessageListener$EndpointMessageListenerAsyncCallback.done(EndpointMessageListener.java:196)
at org.apache.camel.component.jms.EndpointMessageListener.onMessage(EndpointMessageListener.java:117)
at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:719)
at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:679)
at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:649)
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:317)
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:255)
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1168)
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1160)
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1057)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 5000 millis due reply message with correlationID: Camel-ID-ITEM-S69138-1565163898705-0-2 not received on destination: temp-queue://ID:ITEM-S69138-55901-1565163899733-4:20:1. Exchange[ID-ITEM-S69138-1565163898705-0-1]
at org.apache.camel.component.jms.reply.ReplyManagerSupport.processReply(ReplyManagerSupport.java:170)
at org.apache.camel.component.jms.reply.TemporaryQueueReplyHandler.onTimeout(TemporaryQueueReplyHandler.java:62)
at org.apache.camel.component.jms.reply.CorrelationTimeoutMap$1.run(CorrelationTimeoutMap.java:60)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
... 3 more
2019-08-07 09:45:11,997 [Camel (camel-1) thread #25 - JmsReplyManagerOnTimeout[q2]] ERROR DefaultErrorHandler - Failed delivery for (MessageId: ID:ITEM-S69138-55901-1565163899733-4:19:1:1:1 on ExchangeId: ID-ITEM-S69138-1565163898705-0-1). Exhausted after delivery attempt: 1 caught: org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 5000 millis due reply message with correlationID: Camel-ID-ITEM-S69138-1565163898705-0-2 not received on destination: temp-queue://ID:ITEM-S69138-55901-1565163899733-4:20:1. Exchange[ID-ITEM-S69138-1565163898705-0-1]
Message History
---------------------------------------------------------------------------------------------------------------------------------------
RouteId ProcessorId Processor Elapsed (ms)
[route1 ] [route1 ] [jms://queue:q1 ] [ 9051]
[route1 ] [setExchangePattern] [setExchangePattern[InOut] ] [ 6]
[route1 ] [setHeader1 ] [setHeader[CamelJmsRequestTimeout] ] [ 0]
[route1 ] [process1 ] [Processor@0x39652a30 ] [ 3001]
[route1 ] [to1 ] [log:input ] [ 4]
[route1 ] [to2 ] [jms:queue:q2 ] [ 6019]
Stacktrace
---------------------------------------------------------------------------------------------------------------------------------------
org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 5000 millis due reply message with correlationID: Camel-ID-ITEM-S69138-1565163898705-0-2 not received on destination: temp-queue://ID:ITEM-S69138-55901-1565163899733-4:20:1. Exchange[ID-ITEM-S69138-1565163898705-0-1]
at org.apache.camel.component.jms.reply.ReplyManagerSupport.processReply(ReplyManagerSupport.java:170)
at org.apache.camel.component.jms.reply.TemporaryQueueReplyHandler.onTimeout(TemporaryQueueReplyHandler.java:62)
at org.apache.camel.component.jms.reply.CorrelationTimeoutMap$1.run(CorrelationTimeoutMap.java:60)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
2019-08-07 09:45:12,062 [Camel (camel-1) thread #3 - JmsConsumer[q3]] INFO input - Exchange[ExchangePattern: InOut, BodyType: String, Body: stuff in the body]
2019-08-07 09:45:12,093 [Camel (camel-1) thread #4 - JmsConsumer[q4]] INFO input - Exchange[ExchangePattern: InOnly, BodyType: String, Body: stuff in the body]
2019-08-07 09:45:15,034 [Camel (camel-1) thread #28 - JmsReplyManagerOnTimeout[q3]] WARN TemporaryQueueReplyManager - Timeout occurred after 5000 millis waiting for reply message with correlationID [Camel-ID-ITEM-S69138-1565163898705-0-4] on destination temp-queue://ID:ITEM-S69138-55901-1565163899733-4:21:1. Setting ExchangeTimedOutException on (MessageId: ID:ITEM-S69138-55901-1565163899733-4:1:1:1:1 on ExchangeId: ID-ITEM-S69138-1565163898705-0-3) and continue routing.
2019-08-07 09:45:15,038 [Camel (camel-1) thread #2 - JmsConsumer[q2]] WARN EndpointMessageListener - Execution of JMS message listener failed. Caused by: [org.apache.camel.RuntimeCamelException - org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 5000 millis due reply message with correlationID: Camel-ID-ITEM-S69138-1565163898705-0-4 not received on destination: temp-queue://ID:ITEM-S69138-55901-1565163899733-4:21:1. Exchange[ID-ITEM-S69138-1565163898705-0-3]]
org.apache.camel.RuntimeCamelException: org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 5000 millis due reply message with correlationID: Camel-ID-ITEM-S69138-1565163898705-0-4 not received on destination: temp-queue://ID:ITEM-S69138-55901-1565163899733-4:21:1. Exchange[ID-ITEM-S69138-1565163898705-0-3]
at org.apache.camel.util.ObjectHelper.wrapRuntimeCamelException(ObjectHelper.java:1830)
at org.apache.camel.component.jms.EndpointMessageListener$EndpointMessageListenerAsyncCallback.done(EndpointMessageListener.java:196)
at org.apache.camel.component.jms.EndpointMessageListener.onMessage(EndpointMessageListener.java:117)
at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:719)
at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:679)
at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:649)
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:317)
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:255)
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1168)
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1160)
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1057)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 5000 millis due reply message with correlationID: Camel-ID-ITEM-S69138-1565163898705-0-4 not received on destination: temp-queue://ID:ITEM-S69138-55901-1565163899733-4:21:1. Exchange[ID-ITEM-S69138-1565163898705-0-3]
at org.apache.camel.component.jms.reply.ReplyManagerSupport.processReply(ReplyManagerSupport.java:170)
at org.apache.camel.component.jms.reply.TemporaryQueueReplyHandler.onTimeout(TemporaryQueueReplyHandler.java:62)
at org.apache.camel.component.jms.reply.CorrelationTimeoutMap$1.run(CorrelationTimeoutMap.java:60)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
... 3 more
2019-08-07 09:45:15,039 [Camel (camel-1) thread #28 - JmsReplyManagerOnTimeout[q3]] ERROR DefaultErrorHandler - Failed delivery for (MessageId: ID:ITEM-S69138-55901-1565163899733-4:1:1:1:1 on ExchangeId: ID-ITEM-S69138-1565163898705-0-3). Exhausted after delivery attempt: 1 caught: org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 5000 millis due reply message with correlationID: Camel-ID-ITEM-S69138-1565163898705-0-4 not received on destination: temp-queue://ID:ITEM-S69138-55901-1565163899733-4:21:1. Exchange[ID-ITEM-S69138-1565163898705-0-3]
Message History
---------------------------------------------------------------------------------------------------------------------------------------
RouteId ProcessorId Processor Elapsed (ms)
[route2 ] [route2 ] [jms://queue:q2 ] [ 9017]
[route2 ] [setExchangePattern] [setExchangePattern[InOut] ] [ 0]
[route2 ] [setHeader2 ] [setHeader[CamelJmsRequestTimeout] ] [ 0]
[route2 ] [process2 ] [Processor@0x5763a655 ] [ 3001]
[route2 ] [to3 ] [log:input ] [ 2]
[route2 ] [to4 ] [jms:queue:q3 ] [ 6012]
Stacktrace
---------------------------------------------------------------------------------------------------------------------------------------
org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 5000 millis due reply message with correlationID: Camel-ID-ITEM-S69138-1565163898705-0-4 not received on destination: temp-queue://ID:ITEM-S69138-55901-1565163899733-4:21:1. Exchange[ID-ITEM-S69138-1565163898705-0-3]
at org.apache.camel.component.jms.reply.ReplyManagerSupport.processReply(ReplyManagerSupport.java:170)
at org.apache.camel.component.jms.reply.TemporaryQueueReplyHandler.onTimeout(TemporaryQueueReplyHandler.java:62)
at org.apache.camel.component.jms.reply.CorrelationTimeoutMap$1.run(CorrelationTimeoutMap.java:60)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
2019-08-07 09:45:18,071 [Camel (camel-1) thread #29 - JmsReplyManagerOnTimeout[q4]] WARN TemporaryQueueReplyManager - Timeout occurred after 5000 millis waiting for reply message with correlationID [Camel-ID-ITEM-S69138-1565163898705-0-6] on destination temp-queue://ID:ITEM-S69138-55901-1565163899733-4:22:1. Setting ExchangeTimedOutException on (MessageId: ID:ITEM-S69138-55901-1565163899733-4:2:1:1:1 on ExchangeId: ID-ITEM-S69138-1565163898705-0-5) and continue routing.
2019-08-07 09:45:18,072 [Camel (camel-1) thread #3 - JmsConsumer[q3]] WARN EndpointMessageListener - Execution of JMS message listener failed. Caused by: [org.apache.camel.RuntimeCamelException - org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 5000 millis due reply message with correlationID: Camel-ID-ITEM-S69138-1565163898705-0-6 not received on destination: temp-queue://ID:ITEM-S69138-55901-1565163899733-4:22:1. Exchange[ID-ITEM-S69138-1565163898705-0-5]]
org.apache.camel.RuntimeCamelException: org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 5000 millis due reply message with correlationID: Camel-ID-ITEM-S69138-1565163898705-0-6 not received on destination: temp-queue://ID:ITEM-S69138-55901-1565163899733-4:22:1. Exchange[ID-ITEM-S69138-1565163898705-0-5]
at org.apache.camel.util.ObjectHelper.wrapRuntimeCamelException(ObjectHelper.java:1830)
at org.apache.camel.component.jms.EndpointMessageListener$EndpointMessageListenerAsyncCallback.done(EndpointMessageListener.java:196)
at org.apache.camel.component.jms.EndpointMessageListener.onMessage(EndpointMessageListener.java:117)
at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:719)
at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:679)
at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:649)
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:317)
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:255)
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1168)
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1160)
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1057)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 5000 millis due reply message with correlationID: Camel-ID-ITEM-S69138-1565163898705-0-6 not received on destination: temp-queue://ID:ITEM-S69138-55901-1565163899733-4:22:1. Exchange[ID-ITEM-S69138-1565163898705-0-5]
at org.apache.camel.component.jms.reply.ReplyManagerSupport.processReply(ReplyManagerSupport.java:170)
at org.apache.camel.component.jms.reply.TemporaryQueueReplyHandler.onTimeout(TemporaryQueueReplyHandler.java:62)
at org.apache.camel.component.jms.reply.CorrelationTimeoutMap$1.run(CorrelationTimeoutMap.java:60)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
... 3 more
2019-08-07 09:45:18,073 [Camel (camel-1) thread #29 - JmsReplyManagerOnTimeout[q4]] ERROR DefaultErrorHandler - Failed delivery for (MessageId: ID:ITEM-S69138-55901-1565163899733-4:2:1:1:1 on ExchangeId: ID-ITEM-S69138-1565163898705-0-5). Exhausted after delivery attempt: 1 caught: org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 5000 millis due reply message with correlationID: Camel-ID-ITEM-S69138-1565163898705-0-6 not received on destination: temp-queue://ID:ITEM-S69138-55901-1565163899733-4:22:1. Exchange[ID-ITEM-S69138-1565163898705-0-5]
Message History
---------------------------------------------------------------------------------------------------------------------------------------
RouteId ProcessorId Processor Elapsed (ms)
[route3 ] [route3 ] [jms://queue:q3 ] [ 9010]
[route3 ] [setExchangePattern] [setExchangePattern[InOut] ] [ 0]
[route3 ] [setHeader3 ] [setHeader[CamelJmsRequestTimeout] ] [ 0]
[route3 ] [process3 ] [Processor@0x40c8067 ] [ 3000]
[route3 ] [to5 ] [log:input ] [ 0]
[route3 ] [to6 ] [jms:queue:q4 ] [ 6009]
Stacktrace
---------------------------------------------------------------------------------------------------------------------------------------
org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 5000 millis due reply message with correlationID: Camel-ID-ITEM-S69138-1565163898705-0-6 not received on destination: temp-queue://ID:ITEM-S69138-55901-1565163899733-4:22:1. Exchange[ID-ITEM-S69138-1565163898705-0-5]
at org.apache.camel.component.jms.reply.ReplyManagerSupport.processReply(ReplyManagerSupport.java:170)
at org.apache.camel.component.jms.reply.TemporaryQueueReplyHandler.onTimeout(TemporaryQueueReplyHandler.java:62)
at org.apache.camel.component.jms.reply.CorrelationTimeoutMap$1.run(CorrelationTimeoutMap.java:60)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
【问题讨论】：
-
我不完全理解你的问题。你在
q1上得到一个ExchangeTimedOutException，这是正确的，因为处理时间超过5 秒。发送太晚的“回复”是什么意思？ -
根据我的例子，CamelJmsRequestTimeout设置为5s，一个处理器的处理时间（px）为3s。我想在处理器 p1 之后直接发送对 q1 的回复。但是对 q1 的回复是在 p3 处理之后发送的。所以，如果每个处理器耗时3s，当处理将在p2处理时（p2处理~2s + p1处理3sec = 5s后）就会出现q1的超时
-
这是我理解的部分。看不懂的是
Why the reply to q1 is not sent after p1 instead, manifestly, after p3?这句p3后面到底发回了什么？ExchangeTimedOutException还是处理结果？ -
发送的是放在临时队列（由骆驼自动创建）中的回复，以确认队列消耗的数据。
-
编辑：我添加了实际执行的结果
标签： java apache-camel jms