java - Camel /ActiveMQ交易，重新交付和DLQ

使用Fabric8 379构建。

当前正在与ActiveMQ和Camel挣扎，以使TransactionErrorHandler的预期行为能够按预期工作。

首先根据Camel错误处理程序文档（http://camel.apache.org/error-handler.html），如果我按照建议的方式调用TransactionErrorHandler，即

<errorHandler id="txEH" type="TransactionErrorHandler">
<redeliveryPolicy logStackTrace="false" logExhausted="false" maximumRedeliveries="3"/>
</errorHandler>

我收到一个错误：

Caused by: org.xml.sax.SAXParseException: cvc-enumeration-valid: Value 'TransactionErrorHandler' is not facet-valid with respect to enumeration '[DeadLetterChannel, DefaultErrorHandler, NoErrorHandler, LoggingErrorHandler]'. It must be a value from the enumeration.

哪一个公平，我猜想TransactionErrorHandler已从架构中删除，必须以不同的方式调用？因此，如果我走替代路线并像下面这样指定一个TransactionErrorHandler bean：

<bean id="transactionErrorHandler"
    class="org.apache.camel.spring.spi.TransactionErrorHandlerBuilder">
    <property name="deadLetterUri" value="activemq:queue:ActiveMQ.DLQ" />
    <property name="redeliveryPolicy" ref="redeliveryPolicy" />
</bean>

<bean id="redeliveryPolicy" class="org.apache.camel.processor.RedeliveryPolicy">
    <property name="backOffMultiplier" value="2" />
    <property name="maximumRedeliveries" value="2" />
    <property name="redeliveryDelay" value="1000" />
    <property name="useExponentialBackOff" value="true" />
</bean>

通过指定errorHandlerRef =“ transactionErrorHandler”，我可以在路由中成功使用它。但是，在测试此方案时，将完全忽略redeliveryPolicy，重新交付尝试次数为6（默认），而不是上面指定的2。我希望有人可以指出正确的方向，围绕如何正确指定路由内的TransactionErrorHandler。下面是我当前的测试blueprint.xml，它已部署到结构上：

<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:cm="http://aries.apache.org/blueprint/xmlns/blueprint-cm/v1.0.0"
    xmlns:camel="http://camel.apache.org/schema/blueprint"
    xsi:schemaLocation="http://www.osgi.org/xmlns/blueprint/v1.0.0 http://www.osgi.org/xmlns/blueprint/v1.0.0/blueprint.xsd
                              http://camel.apache.org/schema/blueprint http://camel.apache.org/schema/blueprint/camel-blueprint.xsd
                              http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd
                              http://aries.apache.org/blueprint/xmlns/blueprint-cm/v1.0.0 http://aries.apache.org/schemas/blueprint-cm/blueprint-cm-1.0.0.xsd">

    <!-- blueprint property placeholders -->
    <cm:property-placeholder id="test-adapter" persistent-id="uk.test.transactions">
        <cm:default-properties>
            <cm:property name="amqBrokerURL" value="discovery:(fabric:platform)" />
            <cm:property name="amqBrokerUserName" value="admin" />
            <cm:property name="amqBrokerPassword" value="admin" />
        </cm:default-properties>
    </cm:property-placeholder>

    <camelContext xmlns="http://camel.apache.org/schema/blueprint" id="TestRouteContext" useMDCLogging="true">

        <!-- <errorHandler id="txEH" type="TransactionErrorHandler">
            <redeliveryPolicy logStackTrace="false"
                logExhausted="false" />
        </errorHandler> -->

        <route id="platform-test-route" errorHandlerRef="txEH">
            <from uri="activemq:queue:test-queue-in" />
            <transacted ref="transactionPolicy" />
                    <!-- Basic Bean that logs a message -->
            <bean ref="stubSuccess" />
                    <!-- Basic Bean that throws a java.lang.Exception-->
            <bean ref="stubFailure" />
            <to uri="activemq:queue:test-queue-out" />
        </route>

    </camelContext>

    <bean id="stubSuccess" class="uk.test.transactions.stubs.StubSuccess" />

    <bean id="stubFailure" class="uk.test.transactions.stubs.StubFailure" />

    <bean id="transactionErrorHandler"
        class="org.apache.camel.spring.spi.TransactionErrorHandlerBuilder">
        <property name="deadLetterUri" value="activemq:queue:ActiveMQ.DLQ" />
        <property name="redeliveryPolicy" ref="redeliveryPolicy" />
    </bean>

    <bean id="transactionPolicy" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
        <property name="transactionManager" ref="jmsTransactionManager" />
        <property name="propagationBehaviorName" value="PROPAGATION_REQUIRED" />
    </bean>

    <bean id="jmsTransactionManager"
        class="org.springframework.jms.connection.JmsTransactionManager">
        <property name="connectionFactory" ref="jmsPooledConnectionFactory" />
    </bean>

    <bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
        <property name="connectionFactory" ref="jmsPooledConnectionFactory" />
        <property name="transacted" value="true" />
        <property name="transactionManager" ref="jmsTransactionManager" />
        <property name="cacheLevelName" value="CACHE_CONSUMER" />
    </bean>

    <bean id="jmsPooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory"
        init-method="start" destroy-method="stop">
        <property name="maxConnections" value="1" />
        <property name="connectionFactory" ref="jmsConnectionFactory" />
    </bean>

<!--    <bean id="redeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
        <property name="backOffMultiplier" value="2" />
        <property name="initialRedeliveryDelay" value="2000" />
        <property name="maximumRedeliveries" value="2" />
        <property name="redeliveryDelay" value="1000" />
        <property name="useExponentialBackOff" value="true" />
    </bean> -->

    <bean id="redeliveryPolicy" class="org.apache.camel.processor.RedeliveryPolicy">
        <property name="backOffMultiplier" value="2" />
        <property name="maximumRedeliveries" value="2" />
        <property name="redeliveryDelay" value="1000" />
        <property name="useExponentialBackOff" value="true" />
    </bean>

    <bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="${amqBrokerURL}" />
        <property name="userName" value="${amqBrokerUserName}" />
        <property name="password" value="${amqBrokerPassword}" />
        <property name="watchTopicAdvisories" value="false" />
        <!-- <property name="redeliveryPolicy" ref="redeliveryPolicy" /> -->
    </bean>

</blueprint>

如果有人看到我要去哪里错了，将不胜感激。

最佳答案

您应该在AMQ代理上配置重新交付选项，就像在使用TX时一样，其代理负责重新交付（不是骆驼）。