如何在RabbitMQ消息处理失败时自动添加拒签时间戳至消息头？

可用的拦截器抽象 #

Spring AMQP 提供了 RabbitListenerErrorHandler 接口，用于统一处理 @RabbitListener 方法抛出的异常，无需修改每个消息处理器的业务代码，完全适配你不想大量修改代码和配置的需求。同时可通过配置全局监听容器工厂，让所有消息监听实例自动应用该错误处理器。

实现示例 #

1. 自定义全局错误处理器 #

创建实现 RabbitListenerErrorHandler 的类，在异常处理时为消息添加异常时间戳头，随后保留原拒签逻辑让消息进入死信队列：

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.stereotype.Component;

import java.time.Instant;
import java.util.Map;

@Component
public class TimestampAddingErrorHandler implements RabbitListenerErrorHandler {

    private static final String ERROR_TIMESTAMP_HEADER = "error-timestamp";

    @Override
    public Object handleError(Message message, ListenerExecutionFailedException exception) throws Exception {
        // 为消息头添加异常发生的毫秒级Unix时间戳
        Map<String, Object> headers = message.getMessageProperties().getHeaders();
        headers.put(ERROR_TIMESTAMP_HEADER, Instant.now().toEpochMilli());

        // 抛出原异常，让容器按原有逻辑拒签消息，进入死信队列
        throw exception;
    }
}

2. 配置全局监听容器工厂 #

修改Spring Boot的RabbitMQ监听容器工厂配置，将自定义错误处理器设置为全局默认，所有@RabbitListener都会自动使用该处理器：

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class RabbitMQConfig {

    @Autowired
    private TimestampAddingErrorHandler errorHandler;

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        // 绑定全局错误处理器
        factory.setErrorHandler(errorHandler);
        return factory;
    }

    // 保留你原有的队列配置
    private static final String X_DEAD_LETTER_EXCHANGE = "x-dead-letter-exchange";
    private static final String X_DEAD_LETTER_ROUTING_KEY = "x-dead-letter-routing-key";
    private static final String ERRORS_POSTFIX = ".errors";
    private String myName = "my-queue"; // 替换为你的实际队列名称

    @Bean
    public Queue myQueue() {
        Map<String, Object> parameters = new HashMap<>();
        parameters.put(X_DEAD_LETTER_EXCHANGE, "");
        parameters.put(X_DEAD_LETTER_ROUTING_KEY, myName + ERRORS_POSTFIX);
        return new Queue(myName, true, false, false, parameters);
    }

    @Bean
    public Queue myErrorQueue(Queue myQueue) {
        return new Queue(myQueue.getName() + ERRORS_POSTFIX, true, false, false);
    }
}

3. 验证效果 #

当消息处理抛出异常时，TimestampAddingErrorHandler会自动为消息添加error-timestamp头，随后消息被拒签并进入对应的死信队列。你可在死信队列的消息头中读取该时间戳，用于关联对应日志文件。

关键说明 #

• 该方案无需修改任何现有消息处理器的业务代码，仅需添加全局错误处理器和配置容器工厂，对现有代码侵入性极低。
• 时间戳采用毫秒级Unix格式，便于日志系统解析和关联。
• 完全保留原有死信队列路由逻辑，无需调整队列配置。

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