RabbitMQ集成Spring Integration：QueueChannel移交后WorkerFlow错误处理方案

你的推测完全正确！当消息从Amqp.inboundAdapter流转到QueueChannel时，默认已经完成了自动确认（auto-ack）——这意味着RabbitMQ已经认为消息被成功消费，后续WorkerFlow里的异常根本不会触发AMQP层面的重试逻辑，自然也不会被转发到死信交换机（DLX）。

要解决这个问题，我们有两种清晰的方案，你可以根据自己的需求选择：

方案一：给WorkerFlow添加本地重试+死信转发 #

这种方案不需要改动AMQP入站适配器的配置，直接在WorkerFlow的消息处理环节包裹重试逻辑，复用你已经定义好的retryAdvice和RepublishMessageRecoverer，改动最小。

只需要修改workerFlow方法，在每个handle处理器上添加adviceChain配置：

private IntegrationFlow workerFlow(QueueChannel channel) {
    return IntegrationFlows
            .from(channel)
            .<Object, Class<?>>route(Object::getClass, m -> m
                    .resolutionRequired(true)
                    .defaultOutputToParentFlow()
                    .subFlowMapping(EventOne.class, s -> s.handle(oneHandler, 
                            handlerSpec -> handlerSpec.adviceChain(retryAdvice()))) // 绑定重试Advice
                    .subFlowMapping(EventTwo.class, s -> s.handle(anotherHandler, 
                            handlerSpec -> handlerSpec.adviceChain(retryAdvice())))
            )
            .get();
}

原理说明：

• 当WorkerFlow中的处理器抛出异常时，RetryAdvice会立即触发本地重试（基于你配置的指数退避策略），重试期间消息会保留在内存的QueueChannel中。
• 重试耗尽后，RepublishMessageRecoverer会自动把消息转发到你指定的error.exchange.dlx死信交换机，完全符合你的需求。

方案二：延迟AMQP消息确认，直到Worker处理完成 #

如果你希望把重试和死信逻辑完全统一到RabbitMQ层面（让失败消息先回到原队列重试，再进入DLX），可以调整入站适配器的确认模式，延迟确认时机到WorkerFlow处理完成后。

步骤1：修改eventConsumerFlow的入站适配器配置

将消息确认模式改为MANUAL，让RabbitMQ等待手动确认后才标记消息为已消费：

@Bean
public IntegrationFlow eventConsumerFlow(RabbitTemplate rabbitTemplate, Advice retryAdvice) {
    return IntegrationFlows
            .from(
                    Amqp.inboundAdapter(new SimpleMessageListenerContainer(rabbitTemplate.getConnectionFactory()))
                            .configureContainer(c -> c
                                    .adviceChain(retryAdvice())
                                    .addQueueNames(queueNames)
                                    .prefetchCount(amqpProperties.getPreMatch().getDefinition().getQueues().getEvent().getPrefetch())
                                    .acknowledgeMode(AcknowledgeMode.MANUAL) // 设置手动确认
                            )
                            .messageConverter(rabbitTemplate.getMessageConverter())
                            .acknowledgeMode(AcknowledgeMode.MANUAL) // 同步设置适配器的确认模式
            )
            .<Event, String>route(e -> String.format("worker-input-%d", e.getId() % numberOfWorkers))
            .get();
}

步骤2：在WorkerFlow中添加手动确认逻辑

在每个处理器执行完成后，手动确认消息；如果抛出异常，则拒绝消息（让其进入DLX，前提是原队列已配置DLX）：

private IntegrationFlow workerFlow(QueueChannel channel) {
    return IntegrationFlows
            .from(channel)
            .<Object, Class<?>>route(Object::getClass, m -> m
                    .resolutionRequired(true)
                    .defaultOutputToParentFlow()
                    .subFlowMapping(EventOne.class, s -> s.handle(oneHandler)
                            .handle((payload, headers) -> {
                                // 获取AMQP通道和投递标签
                                Channel channel = headers.get(AmqpHeaders.CHANNEL, Channel.class);
                                Long deliveryTag = headers.get(AmqpHeaders.DELIVERY_TAG, Long.class);
                                try {
                                    // 处理成功，手动确认消息
                                    channel.basicAck(deliveryTag, false);
                                } catch (IOException e) {
                                    throw new RuntimeException("Failed to acknowledge message", e);
                                }
                                return payload;
                            }))
                    .subFlowMapping(EventTwo.class, s -> s.handle(anotherHandler)
                            .handle((payload, headers) -> {
                                // 同样的确认逻辑
                                Channel channel = headers.get(AmqpHeaders.CHANNEL, Channel.class);
                                Long deliveryTag = headers.get(AmqpHeaders.DELIVERY_TAG, Long.class);
                                try {
                                    channel.basicAck(deliveryTag, false);
                                } catch (IOException e) {
                                    throw new RuntimeException("Failed to acknowledge message", e);
                                }
                                return payload;
                            }))
            )
            // 全局异常处理：如果处理器抛出异常，拒绝消息并触发DLX
            .handle((payload, headers) -> {
                Channel channel = headers.get(AmqpHeaders.CHANNEL, Channel.class);
                Long deliveryTag = headers.get(AmqpHeaders.DELIVERY_TAG, Long.class);
                try {
                    // 第三个参数为false，表示不重新入队，直接进入DLX
                    channel.basicNack(deliveryTag, false, false);
                } catch (IOException e) {
                    throw new RuntimeException("Failed to reject message", e);
                }
                return null;
            })
            .get();
}

原理说明：

• 消息进入IntegrationFlow后不会立即确认，直到WorkerFlow处理完成并调用basicAck。
• 如果处理失败，basicNack会拒绝消息，触发RabbitMQ的重试逻辑（你原来的retryAdvice），重试耗尽后消息会被转发到DLX。

方案选择建议 #

• 如果你希望快速实现、改动最小，优先选择方案一：它完全复用你已有的重试和死信配置，不需要和RabbitMQ额外交互，适合轻量场景。
• 如果你需要严格遵循RabbitMQ的重试语义（比如失败消息回到原队列重试），或者希望统一管理所有重试逻辑，再考虑方案二。

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