Spring Cloud Stream：如何重新发布到死信队列并抛出异常答案

【问题标题】：Spring Cloud Stream: how to republish to dead letter queue and also throw exceptionSpring Cloud Stream：如何重新发布到死信队列并抛出异常
【发布时间】：2019-01-18 16:36:45
【问题描述】：

我正在将一个使用 Spring AMQP 的项目迁移到一个使用 Spring Cloud Stream 和 RabbitMQ 的项目。

在我的旧项目中，当使用@RabbitListener 处理消息时发生一些异常时，会引发该异常。如果绑定了死信队列，仍然会抛出异常（如果有重试，只有一次，我猜是最后一次）。这对于记录目的非常有帮助。

在 Spring Cloud 中，如果你定义了属性，@StreamListener 有一个死信队列机制：

spring.cloud.stream.bindings.input1.destination=dest1
spring.cloud.stream.rabbit.bindings.input1.consumer.auto-bind-dlq=true
spring.cloud.stream.rabbit.bindings.input1.consumer.republishToDlq=true

但是如果你有这样的方法（只是一个例子）：

@StreamListener("input1")
public void process(String message){
    System.out.println("Trying...");
    throw new RuntimeException();
}

日志是：

Trying...
Trying...
Trying...
(end of log, no exception thrown)

有没有办法抛出异常（仅在最后一次重试时）？

谢谢！

【问题讨论】：

标签： spring-integration spring-cloud spring-amqp spring-cloud-stream

【解决方案1】：

请参阅有关消费者属性的文档。

设置...consumer.max-attempts=1 禁用重试。

【讨论】：

我想要 1) 重试 2) 重试用尽时的死队列信 3) 抛出最后一个异常，我在使用 Spring AMQP 时拥有所有 3 件事。
嗯，你有spring.cloud.stream.rabbit.bindings.input1.consumer.republishToDlq=true，这意味着绑定器将消息重新发布到 DLQ（并添加异常信息）而不是引发异常，这会导致代理将其发送到 DLQ，但没有异常信息。如果这是您想要的，请删除 republishToDlq，使其默认为 false。
但是如果我将 republishToDql 设置为 false，消息（无一例外）不会在死信队列中重新发布。在文档 (docs.spring.io/spring-cloud-stream/docs/current/reference/…) 中，使用 auto-bind-dlq=true 似乎就足够了，但是如果我删除该属性，消息永远不会到达死信队列。
有两种选择 - 抛出异常，代理将原始消息发送到 DLQ 或 republishToDlq=true，这意味着，binder 将消息发布到 DLQ，并附带额外的诊断信息（异常等在标题）。 auto-bind-dlq 只是设置 DLQ 并配置原始队列以在那里发送拒绝（当抛出异常时）。如果没有 DLQ 配置，队列必须不存在；队列创建后无法更改。
这就是问题所在：“如果没有 DLQ 配置，该队列必须不存在；创建后无法更改队列”。谢谢！！

【解决方案2】：

您可以处理异常，记录它，然后抛出 AmqpRejectAndDontRequeueException。这会将消息发送到死信队列

【讨论】：

【解决方案3】：

您在@StreamListener 下，您希望异常发生在哪里？谁抓住了它？

你可以这样做：

@StreamListener("input1")
public void process(String message){
    try {
        System.out.println("Trying...");
        throw new RuntimeException();
        // or the actual code that handle the message
    } catch (RuntimeException re) {
        // handle the exception, logging etc.
        throw re
    }
}

【讨论】：