Очередь мертвых писем в Camel против ActiveMQ?
Я использую EIP Dead Letter Channel точно так же, как описано в документах Camel для Dead Letter Channel. Вот мой camel.xml (заголовки удалены)
<camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
<!-- this is the Dead Letter Channel error handler, where we send failed message to a log endpoint -->
<route errorHandlerRef="myDeadLetterErrorHandler">
<from uri="jms:q1" />
<doTry>
<validate><simple>${body[subsc]} regex '[a-z0-9]{8}' </simple> </validate>
<choice>
<when>
<simple>${body[key1]} regex '^(v1)$'</simple>
<choice>
<when>
<simple>${body[key2]} == 'v2'</simple>
<to uri="jms:getInfo" />
</when>
</choice>
</when>
<otherwise>
<to uri="jms:invalid" />
<stop />
</otherwise>
</choice>
<doCatch>
<exception>org.apache.camel.ValidationException</exception>
<to uri="jms:invalid"/>
</doCatch>
</doTry>
</route>
<route>
<from uri="jms:queue:deadlc" />
<log message="Got ${body}" loggingLevel="ERROR" logName="cool" />
</route>
</camelContext>
<!--
Lets configure some Camel endpoints
http://camel.apache.org/components.html
-->
<!-- configure the camel activemq component to use the current broker -->
<bean id="confact" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://amq-broker?create=false"/>
<property name="userName" value="${activemq.username}"/>
<property name="password" value="${activemq.password}"/>
</bean>
<bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent" >
<property name="connectionFactory" ref="confact" />
</bean>
<bean id="pooledConnectionFactory"
class="org.apache.activemq.pool.PooledConnectionFactory"
init-method="start"
destroy-method="stop">
<property name="maxConnections" value="8" />
<property name="connectionFactory" ref="confact" />
<property name="expiryTimeout" value="-1" />
<property name="idleTimeout" value="-1" />
</bean>
<bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
<property name="connectionFactory" ref="pooledConnectionFactory"/>
<property name="concurrentConsumers" value="8"/>
</bean>
<bean id="jms" class="org.apache.activemq.camel.component.ActiveMQComponent">
<property name="configuration" ref="jmsConfig"/>
</bean>
<bean id="myDeadLetterErrorHandler" class="org.apache.camel.builder.DeadLetterChannelBuilder">
<property name="deadLetterUri" value="jms:queue:deadlc"/>
<property name="redeliveryPolicy" ref="myRedeliveryPolicyConfig"/>
</bean>
<bean id="myRedeliveryPolicyConfig" class="org.apache.camel.processor.RedeliveryPolicy">
<property name="maximumRedeliveries" value="3"/>
<property name="redeliveryDelay" value="5000"/>
</bean>
У меня есть только один маршрут с Content Based Router, который по существу говорит, что если в теле сообщения есть getInfo, тогда маршрут из jms: foo в jms: getInfo, если в теле сообщения есть executeAction, то маршрут из jms.foo в jms: executeAction.
Я ожидаю, что когда потребитель jms: getInfo не запущен, доставка не будет выполнена, попытки повторной доставки будут такими, как указано в весеннем XML. Но ничего не происходит и ничего не попадает в очередь недоставленных сообщений, вместо этого activemq выдает исключение, показанное ниже.
Может кто-нибудь объяснить, почему это не работает? Насколько я понимаю, что когда потребитель (который является пунктом назначения, который получит сообщение от маршрутизатора на основе содержимого) не работает, acivemq знает, что не должно даже доставить его, и должен немедленно выдать исключение, которое затем обрабатывается каналом мертвых писем верблюда. как настроено.
Я использую верблюда 2.10.3 и activemq 5.8.0.
Я подозреваю, что activemq обрабатывает неудачную доставку сообщений, игнорируя конфигурацию верблюжьего XML для очереди недоставленных сообщений, поскольку эти неудачные сообщения отображаются в Activemq.DLQ в веб-консоли activemq. Если потребитель работает, то сообщения успешно доставляются в соответствии с маршрутизацией, указанной в camel.xml.
0: 58: 46,317 | ОТЛАДКА | Отправка JMS-сообщения: queue://getInfo с сообщением: ActiveMQMapMessage {commandId = 0, responseRequired = false, messageId = null, originalDestination = null, originalTransactionId = null, providerId = null, пункт назначения = null,actionId = null, expiration = 0, отметка времени = 0, прибытие = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = a79a0f9e-c88f-4501-a56d-2d5667aa98e0, replyTo = temp-queue://ID:myhost-57639-1382233787365-3:3:6, persistent = false, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, сжатый = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, размер = 0, properties = {breadcrumbId=ID:myhost-57439-1382232345453-1:1:1:1:6, CamelJmsDeliveryMode=1}, readOnlyProperties = false, readOnlyBody = false, droppable = false} ActiveMQMapMessage{ theTable = {cluster=x78} } | org.apache.camel.component.jms.JmsConfiguration | Верблюжья (верблюжья) нить № 2 - JmsConsumer[бар] 20:58:48,044 | ОТЛАДКА | Исключение неактивной записи ID: 8b0d77a3-afef-4612-a49f-22c1b81eb80e (время ожидания превышает 20000 миллисекунд) | org.apache.camel.component.jms.reply.CorrelationTimeoutMap | Верблюжья (верблюжья) нить № 9 - JmsReplyManagerTimeoutChecker[getInfo] 20:58:48,044 | ВНИМАНИЕ | Тайм-аут произошел после 20000 миллисекунд ожидания ответа на сообщение с correlationID [8b0d77a3-afef-4612-a49f-22c1b81eb80e]. Включите ExchangeTimedOutException (MessageId: ID:myhost-57439-1382232345453-1:1:1:1:5 для ExchangeId: ID-myhost-57640-1382233788284-0-2) и продолжайте маршрутизацию. | org.apache.camel.component.jms.reply.TeilitaryQueueReplyManager | Верблюжья (верблюжья) нить #9 - JmsReplyManagerTimeoutChecker[getInfo] 20:58:48,045 | ВНИМАНИЕ | Не удалось выполнить прослушиватель сообщений JMS. Вызвано: [org.apache.camel.RuntimeCamelException - org.apache.camel.ExchangeTimedOutException: сообщение OUT не было получено в течение: 20000 миллисекунд, ответное сообщение с correlationID: 8b0d77a3-afef-4612-a49f-22c1b81eb80e не получено. Exchange[JmsMessage[JmsMessageID: ID:myhost-57439-1382232345453-1:1:1:1:5]]] | org.apache.camel.component.jms.EndpointMessageListener | Верблюжий (верблюжий) поток #4 - JmsConsumer[bar] org.apache.camel.RuntimeCamelException: org.apache.camel.ExchangeTimedOutException: сообщение OUT не было получено в течение: 20000 миллисекунд ответное сообщение с correlationID: 8b0d77a3-afef-4612- a49f-22c1b81eb80e не получен. Обмен [JmsMessage [JmsMessageID: ID: myhost-57439-1382232345453-1: 1: 1: 1: 5]] в org.apache.camel.util.ObjectHelper.wrapRuntimeCamelException(ObjectHelper.java:1271) в org.apache.camel.component.jms. (AbstractMessageListenerContainer.java:562) по адресу org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:500) по адресу org.springframework.jms.listener.AjLerGer_Lifeer.Exer_Lifeer.Exer_Lifeer.Exer_Lifeer.Exer_Lifeer.Exer_LifeGer.Exser_Lifeer.Serg.Serg.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:326) в org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:264) в org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1071) в org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1063) в org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:960) в java.util.concurrent.ThreadPoolExecutor$Worker.runTavaExolo.jol $Worker.run(ThreadPoolExecutor.java:908) в java.lang.Thread.run(Thread.java:662), вызванный: org.apache.camel.ExchangeTimedOutException: сообщение OUT не было получено в течение: 20000 миллисекунд ответное сообщение с ID корреляции: 8b0d77a3-afef-4612-a49f-22c1b81eb80e не получен. Обмен [JmsMessage [JmsMessageID: ID: myhost-57439-1382232345453-1: 1: 1: 1: 5]] в org.apache.camel.component.jms.reply.ReplyManagerSupport.processReply(ReplyManagerSupport.java:133) или в.apache.camel.component.jms.reply.TevenQueueReplyHandler.onTimeout(TemporaryQueueReplyHandler.java:61) в org.apache.camel.component.jms.reply.CorrelationTimeoutMap.onEviction(CorrelationTimeout.ap.j.game.jp).component.jms.reply.CorrelationTimeoutMap.onEviction(CorrelationTimeoutMap.java:30) в org.apache.camel.support.DefaultTimeoutMap.purge(DefaultTimeoutMap.java:209) в org.apache.camel.suoutortapap.java:159) в java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:441) в java.util.concurrent.FutureTask$Sync.innerRunAndReset(FutureTask.java:317) в java.util.concurrent.FutureTask.runAndReset(FutureTask.java:150) в java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$101(ScheduledThreadPoolExecu)
1 ответ
Ваше сообщение JMS имеет replyTo = temp-queue://ID:myhost-57639-1382233787365-3:3:6, Camel ожидает получения ответного сообщения, поэтому вы получаете 20000 миллисекунд.