当目的地不可用时如何停止在骆驼路线中消费消息

Question

我正在使用 Camel 桥接 IBM 和 ActiveMQ 队列。

一切正常，但是当目的地由于网络问题而变得不可用时，所有处理的消息都会丢失。

如果目的地不可用，骆驼应该停止从源头消费，并在它可用时重新开始，但我找不到正确的模式/配置来做到这一点。

我正在配置从 IBM MQ 到 IBM MQ 的路由，但我正在寻找一种方法来解决 IBM 和 ActiveMQ 的任意组合。

<bean id="lsywmq" class="org.apache.camel.component.jms.JmsComponent">
    <property name="connectionFactory">
        <bean class="com.ibm.mq.jms.MQQueueConnectionFactory">
            <property name="transportType" value="1" />
            <property name="hostName" value="${mq_lsy_hostname}" />
            <property name="port" value="${mq_lsy_port}" />
            <property name="queueManager" value="${mq_lsy_queuemanager}" />
            <property name="channel" value="${mq_lsy_channel}" />
            <property name="useConnectionPooling" value="true" />
        </bean>
    </property>
</bean>

<bean id="wmq" class="org.apache.camel.component.jms.JmsComponent">
    <property name="connectionFactory">
        <bean class="com.ibm.mq.jms.MQQueueConnectionFactory">
            <property name="transportType" value="1" />
            <property name="hostName" value="${mq_hostname}" />
            <property name="port" value="${mq_port}" />
            <property name="queueManager" value="${mq_queuemanager}" />
            <property name="channel" value="${mq_channel}" />
            <property name="useConnectionPooling" value="true" />
        </bean>
    </property>
</bean>

<camelContext xmlns="http://camel.apache.org/schema/blueprint"
    allowUseOriginalMessage="false" >
    <route id="EFF2AVIO">
        <from uri="lsywmq:queue:{{mq_lsy_queue}}"/>
        <to uri="wmq:queue:{{mq_queue}}"/>
            </route>
</camelContext>

我想让它作为管道工作，如果它卡在目的地，所有管道仍然卡住。

Answer 1

您需要将JMS connectionFactory 的transacted 属性设置为true。
这实际上不会阻止您的消费者，但您将实现 Guaranteed Delivery EIP 模式，因此您的消息不会再丢失。
从现在开始，您还可以为 ActiveMQ connectionFactory 提供redelivery policy，这与您相关，例如好长的redeliveryDelay

您可以像这样使用JMSConfiguration 对象来配置JMS options

<bean id="wmq" class="org.apache.camel.component.jms.JmsComponent">
    <property name="configuration" ref="jmsConfig"/>
</bean>

<bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
    <property name="transacted" value="true"/>
    <property name="connectionFactory" ref="myConnectionFactory"/>
</bean>

<bean id="myConnectionFactory" class="com.ibm.mq.jms.MQQueueConnectionFactory">
    <property name="transportType" value="1" />
    <property name="hostName" value="${mq_lsy_hostname}" />
    <property name="port" value="${mq_lsy_port}" />
    <property name="queueManager" value="${mq_lsy_queuemanager}" />
    <property name="channel" value="${mq_lsy_channel}" />
    <property name="useConnectionPooling" value="true" />
</bean>