問題描述

我正在運(yùn)行一個(gè) Spring AMQP 消息偵聽器.

I have a Spring AMQP message listener running.

public class ConsumerService implements MessageListener {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Override
    public void onMessage(Message message) {
        try {
            testService.process(message); //This process method can throw Business Exception
        } catch (BusinessException e) {
           //Here we can just log the exception. How the retry attempt is made?
        } catch (Exception e) {
           //Here we can just log the exception.  How the retry attempt is made?
        }
    }
}

如您所見，在處理過程中可能會(huì)出現(xiàn)異常.由于 Catch 塊中的特定錯(cuò)誤，我想重試.我無法通過 onMessage 中的異常.如何告訴 RabbitMQ 有異常并重試?

As you can see, there could be exception coming out during process. I want to retry because of a particular error in Catch block. I cannot through exception in onMessage. How to tell RabbitMQ to there is an exception and retry?

推薦答案

由于 onMessage() 不允許拋出已檢查的異常，您可以將異常包裝在 RuntimeException 然后重新扔.

Since onMessage() doesn't allow to throw checked exceptions you can wrap the exception in a RuntimeException and re-throw it.

try {
    testService.process(message);
} catch (BusinessException e) {
    throw new RuntimeException(e);
}

但是請(qǐng)注意，這可能會(huì)導(dǎo)致郵件無限期地重新發(fā)送.以下是它的工作原理:

Note however that this may result in the message to be re-delivered indefinitely. Here is how this works:

RabbitMQ 支持拒絕消息并要求代理重新排隊(duì).這顯示在這里.但是 RabbitMQ 本身并沒有重試策略的機(jī)制，例如設(shè)置最大重試次數(shù)、延遲等.

RabbitMQ supports rejecting a message and asking the broker to requeue it. This is shown here. But RabbitMQ doesn't natively have a mechanism for retry policy, e.g. setting max retries, delay, etc.

使用 Spring AMQP 時(shí)，requeue on reject"是默認(rèn)選項(xiàng).Spring 的 SimpleMessageListenerContainer 默認(rèn)情況下會(huì)在出現(xiàn)未處理的異常時(shí)執(zhí)行此操作.所以在你的情況下，你只需要重新拋出捕獲的異常.但是請(qǐng)注意，如果您無法處理消息并且總是拋出異常，則該異常將無限期地重新傳遞并導(dǎo)致無限循環(huán).

When using Spring AMQP, "requeue on reject" is the default option. Spring's SimpleMessageListenerContainer will by default do this when there is an unhandled exception. So in your case you just need to re-throw the caught exception. Note however that if you cannot process a message and you always throw the exception this will be re-delivered indefinitely and will result in an infinite loop.

您可以通過拋出 AmqpRejectAndDontRequeueException 異常，這種情況下消息不會(huì)被重新排隊(duì).

You can override this behaviour per message by throwing a AmqpRejectAndDontRequeueException exception, in which case the message will not be requeued.

您也可以通過設(shè)置完全關(guān)閉 SimpleMessageListenerContainer 的requeue on reject"行為

You can also switch off the "requeue on reject" behavior of SimpleMessageListenerContainer entirely by setting

container.setDefaultRequeueRejected(false) 

當(dāng)一條消息被拒絕并且沒有重新排隊(duì)時(shí)，如果在 RabbitMQ 中設(shè)置了一個(gè)，它將丟失或轉(zhuǎn)移到 DLQ.

When a message is rejected and not requeued it will either be lost or transferred to a DLQ, if one is set in RabbitMQ.

如果您需要具有最大嘗試、延遲等的重試策略，最簡(jiǎn)單的方法是設(shè)置一個(gè)彈簧無狀態(tài)"RetryOperationsInterceptor，它將在線程內(nèi)進(jìn)行所有重試(使用 Thread.sleep()) 而不會(huì)在每次重試時(shí)拒絕消息(因此每次重試都不會(huì)返回 RabbitMQ).當(dāng)重試用盡時(shí)，默認(rèn)情況下將記錄一個(gè)警告并使用該消息.如果您想發(fā)送到 DLQ，您將需要 RepublishMessageRecoverer 或自定義 MessageRecoverer 拒絕消息無需重新排隊(duì)(在后一種情況下，您還應(yīng)該在隊(duì)列中設(shè)置 RabbitMQ DLQ).使用默認(rèn)消息恢復(fù)器的示例:

If you need a retry policy with max attempts, delay, etc the easiest is to setup a spring "stateless" RetryOperationsInterceptor which will do all retries within the thread (using Thread.sleep()) without rejecting the message on each retry (so without going back to RabbitMQ for each retry). When retries are exhausted, by default a warning will be logged and the message will be consumed. If you want to send to a DLQ you will need either a RepublishMessageRecoverer or a custom MessageRecoverer that rejects the message without requeuing (in that latter case you should also setup a RabbitMQ DLQ on the queue). Example with default message recoverer:

container.setAdviceChain(new Advice[] {
        org.springframework.amqp.rabbit.config.RetryInterceptorBuilder
                .stateless()
                .maxAttempts(5)
                .backOffOptions(1000, 2, 5000)
                .build()
});

這顯然有一個(gè)缺點(diǎn)，即您將在整個(gè)重試期間占用線程.您還可以選擇使用有狀態(tài)的"RetryOperationsInterceptor，它會(huì)在每次重試時(shí)將消息發(fā)送回 RabbitMQ，但延遲仍將通過 Thread.sleep() 實(shí)現(xiàn)code> 在應(yīng)用程序中，加上設(shè)置有狀態(tài)攔截器有點(diǎn)復(fù)雜.

This obviously has the drawback that you will occupy the Thread for the entire duration of the retries. You also have the option to use a "stateful" RetryOperationsInterceptor, which will send the message back to RabbitMQ for each retry, but the delay will still be implemented with Thread.sleep() within the application, plus setting up a stateful interceptor is a bit more complicated.

因此，如果您想在不占用 Thread 的情況下延遲重試，您將需要一個(gè)在 RabbitMQ 隊(duì)列上使用 TTL 的更復(fù)雜的自定義解決方案.如果您不想要指數(shù)退避(因此每次重試時(shí)延遲不會(huì)增加)，它會(huì)更簡(jiǎn)單一些.要實(shí)現(xiàn)這樣的解決方案，您基本上在 rabbitMQ 上創(chuàng)建另一個(gè)帶有參數(shù)的隊(duì)列: "x-message-ttl": <delay time in milliseconds> and "x-dead-letter-exchange":"<原始隊(duì)列的名稱>".然后在主隊(duì)列上設(shè)置 "x-dead-letter-exchange":"".所以現(xiàn)在當(dāng)你拒絕并且不重新排隊(duì)消息時(shí)，RabbitMQ 會(huì)將它重定向到第二個(gè)隊(duì)列.當(dāng) TTL 過期時(shí)，它將被重定向到原始隊(duì)列，從而重新傳遞給應(yīng)用程序.所以現(xiàn)在你需要一個(gè)重試攔截器，它在每次失敗后拒絕發(fā)送給 RabbitMQ 的消息，并跟蹤重試計(jì)數(shù).為了避免需要在應(yīng)用程序中保持狀態(tài)(因?yàn)槿绻膽?yīng)用程序是集群的，則需要復(fù)制狀態(tài))您可以從 RabbitMQ 設(shè)置的 x-death 標(biāo)頭計(jì)算重試次數(shù).在此處查看有關(guān)此標(biāo)頭的更多信息.因此，此時(shí)實(shí)現(xiàn)自定義攔截器比使用這種行為自定義 Spring 有狀態(tài)攔截器更容易.

Therefore, if you want retries with delays without occupying a Thread you will need a much more involved custom solution using TTL on RabbitMQ queues. If you don't want exponential backoff (so delay doesn't increase on each retry) it's a bit simpler. To implement such a solution you basically create another queue on rabbitMQ with arguments: "x-message-ttl": <delay time in milliseconds> and "x-dead-letter-exchange":"<name of the original queue>". Then on the main queue you set "x-dead-letter-exchange":"<name of the queue with the TTL>". So now when you reject and don't requeue a message RabbitMQ will redirect it to the second queue. When TTL expires it will be redirected to the original queue and thus redelivered to the application. So now you need a retry interceptor that rejects the message to RabbitMQ after each failure and also keeps track of the retry count. To avoid the need to keep state in the application (because if your application is clustered you need to replicate state) you can calculate the retry count from the x-death header that RabbitMQ sets. See more info about this header here. So at that point implementing a custom interceptor is easier than customising the Spring stateful interceptor with this behaviour.

還要檢查 Spring AMQP 參考中關(guān)于重試的部分.

這篇關(guān)于Spring異步MessageListener用例發(fā)生業(yè)務(wù)異常時(shí)如何讓RabbitMQ重試的文章就介紹到這了，希望我們推薦的答案對(duì)大家有所幫助，也希望大家多多支持html5模板網(wǎng)！