rabbitmq with spring amqp - 在 AmqpException 的情况下消息卡住 [英] rabbitmq with spring amqp - messages stuck in case of AmqpException

问题描述

我在我的消费者体内抛出了一个 AmqpException.我的期望是消息会以先进先出的顺序返回队列，并在未来的某个时候被重新处理.

I am throwing an AmqpException inside of my consumer. My expectation is that the message will return back to the queue in FIFO order and will be reprocessed sometime in the future.

似乎 Spring AMQP 没有将消息释放回队列.而是尝试一遍又一遍地重新处理失败的消息.这会阻止处理新到达的消息.被卡住的那些在 AMQP 控制台中永远处于解压"状态.

It seems as if Spring AMQP does not release the message back to the queue. But instead tries to reprocess the failed messages over and over again. This blocks the newly arrived messages from being processed. The ones that are stuck appear in the "unpacked" state forever inside of the AMQP console.

有什么想法吗?

推荐答案

rabbitmq/Spring AMQP就是这样工作的；如果一条消息被拒绝(抛出任何异常)，该消息默认重新排队并放回队列的头部，以便立即重试.

That's the way rabbitmq/Spring AMQP works; if a message is rejected (any exception is thrown) the message is requeued by default and is put back at the head of the queue so it is retried immediately.

...在未来某个时候重新处理.

... reprocessed sometime in the future.

您必须进行适当的配置才能做到这一点.

You have to configure things appropriately to make that happen.

首先，您必须告诉代理不要重新排队消息.这是通过将侦听器容器上的 defaultRequeueRejected 设置为 false 来完成的(默认情况下为 true).或者，您可以抛出一个 AmqpRejectAndDontRequeueException 指示容器拒绝(而不是重新排队)单个消息.

First, you have to tell the broker to NOT requeue the message. That is done by setting defaultRequeueRejected on the listener container to false (it's true by default). Or, you can throw an AmqpRejectAndDontRequeueException which instructs the container to reject (and not requeue) an individual message.

但这还没有结束；这样做只会导致被拒绝的消息被丢弃.

But that's not the end of it; just doing that will simply cause the rejected message to be discarded.

为避免这种情况，您必须为queue - 被拒绝的消息然后被发送到 DLX/DLQ 而不是被丢弃.通常建议使用策略而不是队列参数.

To avoid that, you have to set up a Dead Letter Exchange/Queue for the queue - rejected messages are then sent to the DLX/DLQ instead of being discarded. Using a policy rather than queue arguments is generally recommended.

最后，您可以在 DLQ 上设置消息的生存时间，这样，在该时间之后，消息就会从队列中删除.如果您在 那个 队列(DLQ)上设置另一个适当的死信交换，您可以使消息在时间到期后重新排队回到原始队列.

Finally, you can set a message time to live on the the DLQ so, after that time, the message is removed from the queue. If you set up an another appropriate dead letter exchange on that queue (the DLQ), you can cause the message to be requeued back to the original queue after the time expires.

请注意，这仅适用于原始队列中被拒绝的交付；当该队列中的消息过期时，它将不起作用.

Note that this will only work for rejected deliveries from the original queue; it will not work when expiring messages in that queue.

参见 this answer 以及其问题中的一些链接以获取更多详细信息.

See this answer and some of the links from its question for more details.

您可以使用 x-death 标头的内容来决定是否应该在尝试多次后完全放弃(捕获异常并以某种方式处理错误消息；不要抛出一个异常，容器将确认消息).

You can use the contents of the x-death header to decide if you should give up completely after some number of attempts (catch the exception and somehow dispose of the bad message; don't thrown an exception and the container will ack the message).

这篇关于rabbitmq with spring amqp - 在 AmqpException 的情况下消息卡住的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！