WebFlux/响应式Spring RabbitMQ：保存失败时消息仍被确认如何处理？

这个问题的核心在于你用了subscribe()来触发响应式流，而Spring AMQP的@RabbitListener在这种场景下会提前确认消息——因为subscribe()是异步非阻塞的，Listener线程调用后会直接返回，根本不会等待响应式流执行完成。哪怕后续保存失败抛出异常，RabbitMQ已经收到了确认信号，自然会把消息从队列中移除。

要让RabbitMQ正确识别处理失败并将消息转发至死信队列，你需要调整两个关键部分：让Spring AMQP等待响应式流的执行结果，以及正确配置死信队列规则。

1. 修改Listener方法，返回响应式类型 #

Spring AMQP原生支持响应式的Listener方法，只要你返回Mono或Flux，框架就会自动订阅这个流，等待它完成或出错后再进行消息确认/拒绝操作。

修改后的代码示例 #

@RabbitListener(queues = Constants.SOME_QUEUE)
public Mono<Void> receiveMessage(final List<ItemList> itemList) {
    log.info("Received message from queue: {}", Constants.SOME_QUEUE);
    return itemService.saveAll(itemList)
            .doOnNext(item -> log.info("Saving item with {}", item.getId()))
            .doOnError(error -> log.error("Error during saving item", error))
            .doOnComplete(() -> log.info("{} queue - {} items saved", Constants.SOME_QUEUE, itemList.size()))
            .then() // 转换为Mono<Void>，表示流执行完成
            .onErrorMap(error -> new AmqpRejectAndDontRequeueException("Failed to save items", error));
}

关键说明： #

• 移除了手动的subscribe()，改用响应式操作符链式调用，让Spring AMQP接管流的生命周期。
• 当流正常完成时，Spring AMQP会自动发送消息确认信号给RabbitMQ。
• 当流抛出异常时（这里通过onErrorMap把保存异常转换为AmqpRejectAndDontRequeueException），Spring AMQP会拒绝消息，并且不会将其重新入队——这正是触发死信队列转发的前提。

2. 配置死信队列与交换机 #

要让被拒绝的消息进入死信队列，你需要给目标队列（Constants.SOME_QUEUE）配置死信相关参数，同时声明对应的死信交换机和死信队列。

示例配置代码 #

@Configuration
public class RabbitMqConfig {

    // 目标业务队列，配置死信规则
    @Bean
    Queue someQueue() {
        return QueueBuilder.durable(Constants.SOME_QUEUE)
                // 指定死信交换机
                .withArgument("x-dead-letter-exchange", Constants.DLX_EXCHANGE)
                // 指定死信消息的路由键（可选，若不设置则使用原消息的路由键）
                .withArgument("x-dead-letter-routing-key", Constants.DLQ_ROUTING_KEY)
                // 可选：设置消息超时时间，超时后自动进入死信队列
                .withArgument("x-message-ttl", 30000)
                .build();
    }

    // 死信交换机
    @Bean
    DirectExchange dlxExchange() {
        return new DirectExchange(Constants.DLX_EXCHANGE);
    }

    // 死信队列
    @Bean
    Queue dlqQueue() {
        return QueueBuilder.durable(Constants.DLQ_QUEUE).build();
    }

    // 绑定死信队列到死信交换机
    @Bean
    Binding dlqBinding() {
        return BindingBuilder.bind(dlqQueue())
                .to(dlxExchange())
                .with(Constants.DLQ_ROUTING_KEY);
    }
}

3. 额外注意事项 #

• 不需要设置returnExceptions = "true"，因为返回响应式类型时，框架会自动处理异常的传播。
• 如果你的itemService.saveAll()本身已经会抛出业务异常，也可以不用onErrorMap转换，直接让异常 propagate——Spring AMQP默认会拒绝消息并不重新入队，但显式转换为AmqpRejectAndDontRequeueException可以更明确地控制行为。

内容的提问来源于stack exchange，提问作者justAnotherDeveloper