在未来的某个时间点重试消息 (ActiveMQ) [英] Retrying messages at some point in the future (ActiveMQ)

问题描述

我正在 ActiveMQ 中处理一个系统，我真的不希望丢失消息.我的问题是重试消息导致我的消费者阻塞(而不是处理他们可以处理的消息).我想给失败的消息几天重试(例如，我的潜在目的地之一是我将通过 SFTP 访问的另一台服务器，该服务器可能已关闭)，但我不希望消费者阻塞几天 -- 我希望它继续处理其他消息.

I'm working on a system in ActiveMQ where I would really prefer not to lose messages. My problem is that retrying messages is causing my consumers to block (instead of working on messages they could handle). I'd like to give failed messages several days to retry (for example, one of my potential destinations is another server I'll access via SFTP, which could be down), but I don't want a consumer blocking for several days -- I want it to keep working on the other messages.

有没有办法告诉代理稍后重新发送消息?现在我正在考虑将消息从队列中取出并延迟放置，但我想知道是否有更简单的方法.我正在使用 Apache Camel，所以使用它的解决方案也很好.

Is there a way to tell the broker to resend the message later? Right now I'm looking into taking the message off of the queue and putting it on with a delay, but I'm wondering if there's a simpler way. I'm using Apache Camel, so a solution using that would be good too.

推荐答案

Camel 绝对可以帮助解决这个问题...

Camel can definitely help with this...

一种方法是使用单独的队列并定期重试与主流分开的消息(尤其是在性能受到关注时).此外，这提供了一个单独的队列，允许您对这些错误消息进行分类(查看、清除、更改、手动重试等)...

One way is to use a separate queue and periodically retry messages separately from the main flow (especially when performance is a concern). Also, this provides a separate queue to allow you to triage those error messages (view, clear, alter, manually retry, etc)...

类似这样的事情...请参阅轮询消费者了解更多详情

something like this...see polling consumer for more details

//main route to process message from a queue (needs to be fast)
from("activemq:queue:mainQ").process(...);

//handle any errors by simply moving them to an error queue (for retry later)
onException(Exception.class)
    .handled(true).to("activemq:queue:mainErrorQ");

//retry the error queue
from("timer://retryTimer?fixedRate=true&period=60000")
    .bean(myBean, "retryErrors"); 

...

public void retryErrors() {
    // loop to empty queue
    while (true) {
        // receive the message from the queue, wait at most 3 sec
        Exchange msg = consumer.receive("activemq:queue.mainErrorQ", 3000);
        if (msg == null) {
            // no more messages in queue
            break;
        }

        // send it to the starting queue
        producer.send("activemq:queue.mainQ", msg);
    }
}   

如果您找到更好的解决方案，请告诉我...祝您好运

If you land on a better solution, let me know...good luck

这篇关于在未来的某个时间点重试消息 (ActiveMQ)的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！