未设置 ActiveMQ RedeliveryPolicy

Question

我正在使用：

• SpringBoot 2.0.4
• ActiveMQ 5.15.5
• Apache Camel 2.22.0
• Java 1.8
• 时髦的
• 马文

基本上，我有一个带有 Apache Camel 路由的 SpringBoot 应用程序，该路由使用来自 ActiveMQ 的带有事务的消息。我需要在 ActiveMQ 上设置一个 RedeliveryPolicy，所以当处理发生错误时，消息会重试多次。

我为 ActiveMQ 制作了一个带有 bean 的配置类，事务按预期工作，但 RedeliveryPolicy 不起作用。有人可以帮我理解这有什么问题吗？

这是产生错误的消息的日志输出：

2018-10-23 10:35:28.005 调试 10524 --- [mer[entryQueue]] oacsspi.TransactionErrorHandler : 事务开始 (0x35d60381) 为 (MessageId: ID:EPIC-LAP-25- 50304-1540306817804-4:3:1:1:2 在 ExchangeId: ID-EPIC-LAP-25-1540312510586-0-1)) 2018-10-23 10:35:28.020 调试 10524 --- [mer[entryQueue]] o.apache.camel.processor.SendProcessor : >>>> direct://middle Exchange[ID-EPIC-LAP-25- 1540312510586-0-1] 2018-10-23 10:35:28.375 DEBUG 10524 --- [mer[entryQueue]] oacamel.processor.DefaultErrorHandler：（MessageId：ID：EPIC-LAP-25-50304-1540306817804-4：3： ExchangeId 上的 1:1:2：ID-EPIC-LAP-25-1540312510586-0-1)。交付尝试：0 捕获：java.lang.RuntimeException：ExceptionTest：订单失败 2018-10-23 10:35:28.390 错误 10524 --- [mer[entryQueue]] oacamel.processor.DefaultErrorHandler：（MessageId：ID：EPIC-LAP-25-50304-1540306817804-4：3： ExchangeId 上的 1:1:2：ID-EPIC-LAP-25-1540312510586-0-1)。交付尝试后用尽：1 捕获：java.lang.RuntimeException: ExceptionTest: Order Failed

这是我的 ActiveMQ 配置类：

import org.apache.activemq.ActiveMQConnectionFactory
import org.apache.activemq.RedeliveryPolicy
import org.apache.activemq.camel.component.ActiveMQComponent
import org.springframework.beans.factory.annotation.Qualifier
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.jms.connection.JmsTransactionManager

import javax.jms.DeliveryMode

@Configuration
class ActiveMQConfiguration {

    @Bean
    ActiveMQConnectionFactory activeMQConnectionFactory() {
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory()
        activeMQConnectionFactory.brokerURL = 'tcp://localhost:61616'
        activeMQConnectionFactory.userName = 'admin'
        activeMQConnectionFactory.password = 'admin'

        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy()
        redeliveryPolicy.maximumRedeliveries = 3
        redeliveryPolicy.redeliveryDelay = 150L
        redeliveryPolicy.useExponentialBackOff = true
        redeliveryPolicy.backOffMultiplier = 1.5

        activeMQConnectionFactory.setRedeliveryPolicy(redeliveryPolicy)

        activeMQConnectionFactory
    }

    @Bean
    ActiveMQComponent activeMQComponent(@Qualifier('activeMQConnectionFactory')ActiveMQConnectionFactory activeMQConnectionFactory) {
        ActiveMQComponent activeMQComponent = new ActiveMQComponent()
        activeMQComponent.connectionFactory = activeMQConnectionFactory
        activeMQComponent.transacted = true
        activeMQComponent.transactionManager = txManager()
        activeMQComponent.cacheLevelName = 'CACHE_CONSUMER'
        activeMQComponent.lazyCreateTransactionManager = false
        activeMQComponent.deliveryMode = DeliveryMode.PERSISTENT

        activeMQComponent
    }

    @Bean
    JmsTransactionManager txManager(@Qualifier('activeMQConnectionFactory') ActiveMQConnectionFactory activeMQConnectionFactory) {
        JmsTransactionManager txManager = new JmsTransactionManager()
        txManager.connectionFactory = activeMQConnectionFactory
        txManager.rollbackOnCommitFailure = true

        txManager
    }

}

Answer 1

这里有两个问题

1.你有两个事务管理器

由于在 Camel ActiveMQ 组件的配置中有以下两行，您配置了两个事务管理器。这是问题的根源。

activeMQComponent.transacted = true // activates local JMS transactions
activeMQComponent.transactionManager = txManager() // additional tx manager

如果你只是想从 ActiveMQ 消费事务，你不需要配置 Spring 事务管理器。

这两行配置足以通过 ActiveMQ 代理获取本地事务。

activeMQComponent.transacted = true
activeMQComponent.lazyCreateTransactionManager = false

所以你应该删除这一行以及整个txManager bean

activeMQComponent.transactionManager = txManager()

如果您当前在您的骆驼路线中设置了交易标志，您也必须将其删除。正如我所写，即使您删除了所有这些，您从 ActiveMQ 消耗的路由仍然会被处理。

2。无法重新投递

您尚未发布您的 Camel 路由，但根据错误输出，我假设 broker 没有重新交付，因为错误由 Camel 处理。

当错误发生时，Camel 错误处理程序o.a.camel.processor.DefaultErrorHandler 会启动，因为它处理错误，所以消息被提交给代理，因此不会发生重新传递。

尝试禁用 Camel 错误处理以查看代理是否重新传递错误消息。

errorHandler(noErrorHandler());

Answer 2

不久前，我遇到了 dlq 队列的问题 - 并非代码中设置的所有参数都有效。我必须将设置添加到 acitvemq 配置中。是的，划分配置不是一个好的决定，但我没有找到另一个。 下面是我的 jms 配置类和通过 activemq.xml 进行的示例队列配置：

@Configuration
@EnableJms
public class JmsConfig {

    private Environment env;

    @Autowired
    public void setEnv(Environment env) {
        this.env = env;
    }

    @Bean(name = "activemq")
    public ActiveMQComponent activemq(@Qualifier("activemqTransactionManager") JmsTransactionManager jmsTransactionManager,
                                      @Qualifier("activemqConnectionFactory") ConnectionFactory connectionFactory) {
        ActiveMQComponent activeMQComponent = new ActiveMQComponent();
        activeMQComponent.setTransactionManager(jmsTransactionManager);
        activeMQComponent.setConnectionFactory(connectionFactory);
        return activeMQComponent;
    }

    @Bean(name = "activemqJmsTemplate")
    public JmsTemplate jmsTemplate(@Qualifier("activemqConnectionFactory") ConnectionFactory connectionFactory) {
        JmsTemplate template = new JmsTemplate();
        template.setConnectionFactory(connectionFactory);
        return template;
    }

    @Bean(name = "activemqTransactionPolicy")
    public SpringTransactionPolicy activemqTransactionPolicy(
            @Qualifier("activemqTransactionManager") JmsTransactionManager jmsTransactionManager) {
        SpringTransactionPolicy springTransactionPolicy = new SpringTransactionPolicy(jmsTransactionManager);
        springTransactionPolicy.setPropagationBehaviorName("PROPAGATION_REQUIRED");
        return springTransactionPolicy;
    }

    @Bean(name = "activemqTransactionManager")
    public JmsTransactionManager activemqTransactionManager(
            @Qualifier("activemqConnectionFactory") ConnectionFactory connectionFactory) {
        return new JmsTransactionManager(connectionFactory);
    }

    @Bean(name = "activemqConnectionFactory")
    public ConnectionFactory connectionFactory(@Qualifier("activemqRedeliveryPolicy") RedeliveryPolicy redeliveryPolicy) {
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        connectionFactory.setBrokerURL("tcp://" + env.getProperty("queue.url"));
        connectionFactory.setTrustAllPackages(true);

        RedeliveryPolicyMap map = connectionFactory.getRedeliveryPolicyMap();
        map.put(new ActiveMQQueue("queueName.DLQ"), redeliveryPolicy);
        return connectionFactory;
    }

    @Bean(name = "activemqRedeliveryPolicy")
    public RedeliveryPolicy redeliveryPolicy() {
        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        redeliveryPolicy.setMaximumRedeliveries(0);
        return redeliveryPolicy;
    }
}

activevq.xml 的变化：

<destinationPolicy>
    <policyMap>
        <policyEntries>
            <!--set dead letter queue for our queue. It name will be "myQueueName.DLQ"-->
            <policyEntry queue="myQueueName">
                <deadLetterStrategy>
                    <individualDeadLetterStrategy queuePrefix="" queueSuffix=".DLQ"/>
                </deadLetterStrategy>
            </policyEntry>
            <policyEntry topic=">">
                <pendingMessageLimitStrategy>
                    <constantPendingMessageLimitStrategy limit="1000"/>
                </pendingMessageLimitStrategy>
            </policyEntry>
        </policyEntries>
    </policyMap>
</destinationPolicy>

<plugins>
<redeliveryPlugin fallbackToDeadLetter="true" sendToDlqIfMaxRetriesExceeded="true">
    <redeliveryPolicyMap>
        <redeliveryPolicyMap>
            <redeliveryPolicyEntries>
                <!--Set the redelivery delay to one hour-->
                <redeliveryPolicy queue="myQueueName.DLQ" maximumRedeliveries="-1" redeliveryDelay="3600000"/>
            </redeliveryPolicyEntries>
        </redeliveryPolicyMap>
    </redeliveryPolicyMap>
</redeliveryPlugin>
</plugins>