spring-boot - Apache Camel 中的持久重试

Question

我正在努力让 Camel 中的重新传递消息得以保留:

from("activemq:incoming-queue")
     .errorHandler(
         deadLetterChannel("activemq:queue.dead-letter")
         .maximumRedeliveries(100)
         .redeliveryDelay(TimeUnit.SECONDS.toMillis(10))
     )
.to("http4://example.com")

如果发生故障，会安排重试，如果我们关闭 Camel，它会将消息放回到 ActiveMQ 队列中，这很好。

问题是，如果你用“kill -9”杀死它，消息将永远丢失，因为 Camel 在重试期间将它们保存在内存中。

问题是，即使系统突然宕机，如何保证重试时消息不会丢失？

更新:

我们使用交易实现了有保证的重新交付:

    @Bean
    public JmsTransactionManager jmsTransactionManager(ConnectionFactory connectionFactory) {
        JmsTransactionManager jmsTransactionManager = new JmsTransactionManager();
        jmsTransactionManager.setConnectionFactory(connectionFactory);

        return jmsTransactionManager;
    }

    /**
     * Disables ActiveMQ redelivery since it clashes with camel redelivery system
     */
    @Bean
    public RedeliveryPolicy redeliveryPolicy(ActiveMQConnectionFactory connectionFactory) {
        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        redeliveryPolicy.setMaximumRedeliveries(0);
        connectionFactory.setRedeliveryPolicy(redeliveryPolicy);
        return redeliveryPolicy;
    }

然后路由如下所示(注意 transacted=true 参数，由于某些原因 .transacted() 方法的行为不同):

from("activemq:incoming-queue?transacted=true")
         .errorHandler(
             deadLetterChannel("activemq:queue.dead-letter")
             .maximumRedeliveries(100)
             .redeliveryDelay(TimeUnit.SECONDS.toMillis(10))
         )
.to("http4://example.com")

使用这种方法有什么缺点吗？

Answer 1

为了让克劳斯的回答更加具体:

<bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="${activemq.url}"/>
    <property name="userName" value="${activemq.userName}"/>
    <property name="password" value="${activemq.password}"/>
    <property name="redeliveryPolicy">
        <bean class="org.apache.activemq.RedeliveryPolicy">
            <property name="maximumRedeliveries" value="100"/>
            <property name="initialRedeliveryDelay" value="10000"/>
            <property name="redeliveryDelay" value="10000"/>
        </bean>
    </property>
</bean>

使用 ActiveMQ 重试 100 次，第一次重试等待 10 秒，随后每次重试等待 10 秒。

然后你可以省略 Camel 中的错误处理(从而将其留给 ActiveMQ)，但我会添加事务:

from("activemq:incoming-queue")
     .transacted("PROPAGATION_REQUIRED_JMS")
     .to("http4://example.com")

重试 100 次后，消息将最终出现在队列“ActiveMQ.DLQ”中，除非您进行更改。