如何确保来自 JMS 队列的消息被传递到外部 WebService (CXF)？

Question

问题

应该如何在Mule ESB 3.2 中配置ActiveMQ 和<flow>，以确保从队列中拉出的消息最终被外部CXF service 正确处理？

场景

我有一个 CXF 端点，它应该接收传入的消息并尽快将其传输到三个外部服务。我们称它们为 EX1、EX2、EX3。这很容易，这要归功于 Mule 3.x 中引入的 <all> 组件。

整个解决方案最重要的要求，是确保每条收到的消息最终都被传递到所有三个 CXF 服务。所以我们最终有了这个想法，将每条传入的消息放入Persistent JMS queues（Q1、Q2、Q3）。在从队列 Qn 中读取消息后，将其直接传输到相应的 EXn 端点，从而 - 外部服务。

配置

（我可以根据要求提供完整的配置）

我们已经按照here 的描述配置了 ActiveMQ 代理，并将其与我们的 <flow> 配置连接起来。一切似乎都按预期工作，我已将 JConsole 连接到我的应用程序，因此我可以看到消息的类型为 PERSISTENT 并且它们最终出现在正确的队列中。如果一切顺利 - 所有三个服务 EXn 都接收到消息。

测试

当我们关闭其中一项服务（例如 EX2）并重新启动整个服务器以模拟故障时，就会出现问题。 消息最终会丢失（我想这不是那么持久，对吧？）。 最奇怪的是——如果我们在 EX2 宕机时发送了 10 条消息，那么在服务器重新启动后，其中 9 条消息被正确地重新发送了！所以我在想，也许，只是也许，这 10 条消息中有 9 条已正确排队，而当服务器出现故障时，一条正在不断地重新传递。

这让我想到，说实话，我无法理解 CXF 端点没有通过事务支持来处理。毕竟，当它试图重新传递时，我可以看到消息在队列中，所以它应该被持久化。显然不是，但为什么呢？

我自己的尝试 我尝试了很多东西，但都没有奏效。总是有一条消息丢失。

1. 不在流程中使用任何 <jms:transaction /> 标记 - 不起作用
2. 在收到消息时启动 jms 事务，在发送到 <cxf:jaxws-client /> 时加入
3. 将 XA 与 JBoss 和 <xa-transaction /> 一起使用 - 不起作用
4. 提供 <default-exception-strategy> 配置 - 如果我记得它让事情变得更糟

感谢您的帮助。

配置

活动 MQ 配置

<spring:bean id="AmqDefaultPolicyEntry" class="org.apache.activemq.broker.region.policy.PolicyEntry">
    <spring:property name="queue" value="queue.*"/>
    <spring:property name="deadLetterStrategy" ref="AmqDeadLetterStrategy"/>
</spring:bean>

<spring:bean id="AmqPolicyMap" class="org.apache.activemq.broker.region.policy.PolicyMap">
    <spring:property name="defaultEntry" ref="AmqDefaultPolicyEntry"/>
</spring:bean>

<spring:bean name="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory" depends-on="AmqBroker">
    <spring:property name="brokerURL" value="vm://localhost?jms.prefetchPolicy.all=1&amp;broker.persistent=true&amp;broker.useJmx=true"/>
    <spring:property name="redeliveryPolicy">
        <spring:bean class="org.apache.activemq.RedeliveryPolicy">
            <spring:property name="initialRedeliveryDelay" value="${props.initialRedeliveryDelay}"/>
            <spring:property name="redeliveryDelay" value="${props.redeliveryDelay}"/>
            <spring:property name="maximumRedeliveries" value="${props.maximumRedeliveries}"/>
            <spring:property name="backOffMultiplier" value="${props.backOffMultiplier}"/>
        </spring:bean>
    </spring:property>
</spring:bean>

<spring:bean name="persistenceAdapter" class="org.apache.activemq.store.kahadaptor.KahaPersistenceAdapter">
    <spring:property name="directory" value="/home/bachman/activemq"/>
</spring:bean>

<spring:bean name="AmqBroker"
             class="org.apache.activemq.broker.BrokerService"
             init-method="start"
             destroy-method="stop">
    <spring:property name="brokerName" value="esb-amq-broker"/>
    <spring:property name="persistent" value="true"/>
    <spring:property name="dataDirectory" value="/home/bachman/activemq"/>
    <spring:property name="useJmx" value="true"/>
    <spring:property name="useShutdownHook" value="false"/>
    <spring:property name="persistenceAdapter" ref="persistenceAdapter"/>
    <spring:property name="destinationPolicy" ref="AmqPolicyMap"/>
</spring:bean>

<jms:activemq-connector name="PersistentJMSConnector" specification="1.1"
                        numberOfConsumers="1" maxRedelivery="-1" persistentDelivery="true"
                        connectionFactory-ref="connectionFactory" acknowledgementMode="CLIENT_ACKNOWLEDGE"
                        disableTemporaryReplyToDestinations="true"/>

FLOW - 将传入消息分派到 3 个队列 Qn

<flow name="dispatch-to-queues">
        <inbound-endpoint ref="incoming-cxf"/>

        <!-- Each received message ends up to be sent to all destinations -->
        <all>
            <jms:outbound-endpoint name="queue.q1"
                queue="queue.q1" disableTransportTransformer="false"
                disableTemporaryReplyToDestinations="false" exchange-pattern="one-way"
                doc:name="JMS" doc:description="Receive messages on Q1"
                    connector-ref="PersistentJMSConnector"/>

            <jms:outbound-endpoint name="queue.q2"
                queue="queue.q2" disableTransportTransformer="false"
                disableTemporaryReplyToDestinations="false" exchange-pattern="one-way"
                doc:name="JMS" doc:description="Receive messages on q2"
                connector-ref="PersistentJMSConnector" />

            <jms:outbound-endpoint name="queue.q3"
                queue="queue.q3" disableTransportTransformer="false"
                disableTemporaryReplyToDestinations="false" exchange-pattern="one-way"
                doc:name="JMS" doc:description="Receive messages on Q3"
                connector-ref="PersistentJMSConnector" />

        </all>
        <custom-processor class="com.mycompany.just.a.component.to.return.OK.via.Cxf" />
    </flow>

FLOW - 处理从 Qn 到 EXn 的传递

<flow name="from-q1-to-ex1">
        <jms:inbound-endpoint queue="queue.q1" disableTransportTransformer="false"
            disableTemporaryReplyToDestinations="false" exchange-pattern="one-way"
            doc:name="JMS" doc:description="Pull from q1."
            connector-ref="PersistentJMSConnector">
                <jms:transaction action="ALWAYS_BEGIN" />
        </jms:inbound-endpoint>
        <logger message="Sending message to EX-1" level="INFO" />

        <!-- Handle errors at this point in flow
        <custom-processor class="pl.exception.lookup.Component">
            <spring:property name="targetModuleName" value="Not-important"/>
        </custom-processor>
        -->


        <outbound-endpoint ref="ex1-cxf-endpoint">
            <jms:transaction action="ALWAYS_JOIN" timeout="360000"/>
        </outbound-endpoint>
    </flow>

ENDPOINTS - 引用端点的声明

<endpoint name="incoming-cxf" address="http://incoming.mycompany.com/in" exchange-pattern="request-response">
        <cxf:jaxws-service serviceClass="com.mycompany.services.InService"/>
    </endpoint> 

<endpoint name="ex1-cxf-endpoint" address="http://com.mycompany.ex1" exchange-pattern="request-response">
        <cxf:jaxws-client
                clientClass="com.mycompany.services.Ex1"
                wsdlLocation="classpath:wsdl/ex1.wsdl"
                operation="someOperation"
                port="SomePort"/>
    </endpoint>

Answer 1

在事务中使用 JMS 消息是解决方案按预期工作的必要条件：如果在 CXF 出站阶段发生异常，JMS 消息将最终回滚，然后重新传递，触发新的 CXF 调用。

您必须为您的 ActiveMQ 客户端仔细配置重新传递策略，以便重试足够多的时间并且可能不会太快（例如指数回退）。您还想适当地处理 DLQ。在 Mule 中使用 Spring Beans 的 ActiveMQ 客户端配置如图所示：http://www.mulesoft.org/mule-activemq-integration-examples

还要确保在您的配置工厂中引用正确的代理 URL。如果您的代理名称为 esb-amq-broker，您的配置工厂应该是：

<spring:bean name="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory" depends-on="AmqBroker">
    <spring:property name="brokerURL" value="vm://esb-amq-broker"/>
    ...

Answer 2

我不知道我是否会帮助你，但这是关于你的问题的一些建议：

• 您是否尝试过使用不同于 Jboss 提供的事务管理器，我建议使用 Atomikos 进行此类测试
• 像 David 建议的那样，事务似乎是最好的方法，但另一种方法是使用显式确认策略 .... 设置可能很棘手，但类似拦截器的方法可以监视与某些特定端点的连接并发送将 ack 返回到您的 JMS 服务器，这可能很困难，但它肯定会确保消息已正确传递....

祝你好运 高温高压 杰罗姆

Answer 3

不确定这种考虑是否有帮助，但是确认模式呢？是否有可能消息已经传递（在自动确认模式下）但尚未被消费服务端点正确处理？

不知道如何在这种情况下配置显式确认，但可能值得进一步研究。