Spring异步MessageListener用例发生业务异常时如何让

问题描述

我正在运行一个 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?
        }
    }
}

如您所见，在处理过程中可能会出现异常.由于 Catch 块中的特定错误，我想重试.我无法通过 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);
}

但是请注意，这可能会导致邮件无限期地重新发送.以下是它的工作原理:

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

RabbitMQ 支持拒绝消息并要求代理重新排队.这显示在这里.但是 RabbitMQ 本身并没有重试策略的机制，例如设置最大重试次数、延迟等.

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 时，requeue on reject"是默认选项.Spring 的 SimpleMessageListenerContainer 默认情况下会在出现未处理的异常时执行此操作.所以在你的情况下，你只需要重新抛出捕获的异常.但是请注意，如果您无法处理消息并且总是抛出异常，则该异常将无限期地重新传递并导致无限循环.

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 异常，这种情况下消息不会被重新排队.

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

您也可以通过设置完全关闭 SimpleMessageListenerContainer 的requeue on reject"行为

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

container.setDefaultRequeueRejected(false) 

当一条消息被拒绝并且没有重新排队时，如果在 RabbitMQ 中设置了一个，它将丢失或转移到 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.

如果您需要具有最大尝试、延迟等的重试策略，最简单的方法是设置一个弹簧无状态"RetryOperationsInterceptor，它将在线程内进行所有重试(使用 Thread.sleep()) 而不会在每次重试时拒绝消息(因此每次重试都不会返回 RabbitMQ).当重试用尽时，默认情况下将记录一个警告并使用该消息.如果您想发送到 DLQ，您将需要 RepublishMessageRecoverer 或自定义 MessageRecoverer 拒绝消息无需重新排队(在后一种情况下，您还应该在队列中设置 RabbitMQ DLQ).使用默认消息恢复器的示例:

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()
});

这显然有一个缺点，即您将在整个重试期间占用线程.您还可以选择使用有状态的"RetryOperationsInterceptor，它会在每次重试时将消息发送回 RabbitMQ，但延迟仍将通过 Thread.sleep() 实现code> 在应用程序中，加上设置有状态拦截器有点复杂.

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 的情况下延迟重试，您将需要一个在 RabbitMQ 队列上使用 TTL 的更复杂的自定义解决方案.如果您不想要指数退避(因此每次重试时延迟不会增加)，它会更简单一些.要实现这样的解决方案，您基本上在 rabbitMQ 上创建另一个带有参数的队列: "x-message-ttl": <delay time in milliseconds> and "x-dead-letter-exchange":"<原始队列的名称>".然后在主队列上设置 "x-dead-letter-exchange":"".所以现在当你拒绝并且不重新排队消息时，RabbitMQ 会将它重定向到第二个队列.当 TTL 过期时，它将被重定向到原始队列，从而重新传递给应用程序.所以现在你需要一个重试拦截器，它在每次失败后拒绝发送给 RabbitMQ 的消息，并跟踪重试计数.为了避免需要在应用程序中保持状态(因为如果您的应用程序是集群的，则需要复制状态)您可以从 RabbitMQ 设置的 x-death 标头计算重试次数.在此处查看有关此标头的更多信息.因此，此时实现自定义拦截器比使用这种行为自定义 Spring 有状态拦截器更容易.

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 参考中关于重试的部分.

这篇关于Spring异步MessageListener用例发生业务异常时如何让RabbitMQ重试的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持html5模板网！