1

我正在使用camel和activeMq研究重试机制。如果我正在调用的服务器之一关闭，我想要做的是启动重试机制，将请求添加到我的队列中，并每小时将其重新发送到服务器。一切正常，除了我的重试策略似乎被忽略（我的请求在进入我的队列时被重新发送，并且在达到重试次数后它们永远不会添加到 DLQ 中）

我的配置如下所示（从 .cfg 文件中读取值）：

<beans xmlns="http://www.springframework.org/schema/beans"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
<bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="${activemq.broker.url}"/>
    <property name="userName" value="${activemq.broker.username}"/>
    <property name="password" value="${activemq.broker.password}"/>
    <property name="redeliveryPolicy" ref="policy"/>
</bean>

<bean id="policy" class="org.apache.activemq.RedeliveryPolicy">
    <property name="queue" value="*"/>
    <property name="initialRedeliveryDelay" value="${activemq.redelivery.delay}"/>
    <property name="redeliveryDelay" value="${activemq.redelivery.delay}"/>
    <property name="useExponentialBackOff" value="false"/>
    <property name="maximumRedeliveries" value="${activemq.number.of.redeliveries}"/>
</bean>

<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
    <property name="maxConnections" value="8"/>
    <property name="connectionFactory" ref="jmsConnectionFactory"/>
</bean>

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

<bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
    <property name="connectionFactory" ref="pooledConnectionFactory"/>
    <property name="transactionManager" ref="jmsTransactionManager"/>
    <property name="transacted" value="true"/>
    <property name="cacheLevelName" value="CACHE_CONSUMER"/>
    <property name="concurrentConsumers" value="8"/>
</bean>

<bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
    <property name="configuration" ref="jmsConfig"/>
</bean>

调用我的服务的骆驼路线：

<route id="addRegistrationRoute">
        <from uri="direct:addRegistrationRoute"/>

        <to uri="cxf:bean:addRegistrationEndpoint"/>

        <onException>
            <exception>java.net.ConnectException</exception>
            <onWhen>
                <el>${in.headers['previousRoute'] != 'registrationRetryRoute'}</el>
            </onWhen>
            <handled>
                <constant>true</constant>
            </handled>

            <setBody>
                <simple>${headers.request}</simple>
            </setBody>
            <removeHeaders pattern="*"/>
            <to uri="activemq:queue:registrationRetryQueue"/>
            <stop/>
        </onException>
        <onException>
            <exception>org.apache.cxf.interceptor.Fault</exception>
            <onWhen>
                <el>${in.headers['previousRoute'] != 'registrationRetryRoute'}</el>
            </onWhen>
            <handled>
                <constant>true</constant>
            </handled>

            <setBody>
                <simple>${headers.request}</simple>
            </setBody>
            <removeHeaders pattern="*"/>

            <to uri="activemq:queue:registrationRetryQueue"/>
            <stop/>
        </onException>
        <onException>
            <exception>javax.xml.soap.SOAPFault</exception>
            <handled>
                <constant>true</constant>
            </handled>

        </onException>
    </route>
    <route id="registrationRetryRoute">
        <from uri="activemq:queue:registrationRetryQueue"/>

        <setHeader headerName="previousRoute">
            <simple>registrationRetryRoute</simple>
        </setHeader>
        <to uri="direct:addRegistrationRoute"/> <!-- Back to the initial flow. -->
    </route>

如果有人可以告诉我我在配置 activeMq 时做错了什么，我将非常感激！

问候， 罗克珊娜

4

1 回答 1

0

仅当您从消费者（或消息侦听器）抛出错误时，ActiveMQ 才会将消息推送到 DLQ。因此，请尝试捕获您的异常并将其返回，以便相应地启用重试策略。

于 2014-12-20T17:08:20.227 回答