Springboot RabbitMQ 消费失败消息清洗与重试机制

📌 引言

在分布式系统中，RabbitMQ 是一个非常流行的消息队列中间件，广泛用于解耦系统、异步处理任务、提高系统性能。然而，在实际使用中，消费端可能因为 代码异常、数据库故障、网络问题 等原因导致消息消费失败。

如果不对这些失败消息进行处理，可能会导致数据丢失，影响业务流程。因此，我们需要一个 可靠的失败消息清洗和重试机制 来保证消息最终被成功消费或进行合理的存储处理。

📌 1. 失败消息的常见问题

在 RabbitMQ 消费过程中，可能会遇到以下问题：

• 消息处理逻辑异常（如空指针异常、数据格式转换错误）。
• 数据库异常（如主键冲突、唯一索引冲突、数据库连接失败）。
• 外部依赖不可用（如调用第三方 API 超时）。
• 消息重复消费（由于网络抖动或 RabbitMQ 配置问题）。

这些异常可能导致：

• 消息丢失：如果 RabbitMQ 没有进行消息重试，消息可能永远丢失。
• 死循环消费：如果未配置死信队列（DLX），消息可能会不断重新入队，造成死循环。
• 系统压力过大：频繁失败的消息可能会影响正常消息的消费，拖垮整个消费端。

因此，我们需要一个 完整的失败消息清洗与重试机制 来保障系统的稳定性。

📌 2. 解决方案

✅ 目标

1. 记录失败的消息，存入数据库，方便后续排查。
2. 实现自动重试，允许在一定次数内自动重试消费失败的消息。
3. 配置延迟队列，避免立即重试导致的瞬时高并发问题。
4. 超出重试次数后进入死信队列，避免死循环消费，便于后续人工干预。

📝 数据库表设计

我们创建一个 failed_message_log 表来存储失败的消息：

CREATE TABLE failed_message_log (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    message_id VARCHAR(64) NOT NULL,
    exchange_name VARCHAR(255),
    routing_key VARCHAR(255),
    queue_name VARCHAR(255),
    message TEXT NOT NULL,
    error_reason TEXT NOT NULL,
    retry_count INT DEFAULT 0,
    status ENUM('PENDING', 'FAILED', 'RETRYING', 'SUCCESS') DEFAULT 'PENDING',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

字段说明：

• message_id：消息的唯一标识符。
• exchange_name / routing_key / queue_name：记录消息来源，方便追溯。
• message：存储原始消息内容。
• error_reason：记录失败的具体原因。
• retry_count：当前消息的重试次数。
• status：消息状态（PENDING：待处理，FAILED：失败，RETRYING：正在重试，SUCCESS：成功）。
• created_at / updated_at：时间戳，方便查询和统计。

📌 3. 消费端实现

📍 监听 RabbitMQ 消息

@Slf4j // 使用 Lombok 提供的日志功能，简化日志记录
@Component // 让 Spring 管理该组件
public class RabbitMqConsumer {

    private static final int MAX_RETRY_COUNT = 3; // 最大重试次数，超过后将消息放入死信队列

    @Autowired
    private FailedMessageLogMapper failedMessageLogMapper; // 失败消息日志的数据库访问对象，用于记录失败消息

    @Autowired
    private RabbitTemplate rabbitTemplate; // RabbitMQ 消息发送模板，用于消息重试和死信处理

    /**
     * 监听 RabbitMQ 队列，处理消息
     *
     * @param messageId   消息的唯一标识 ID
     * @param redelivered 是否是 RabbitMQ 重新投递的消息（用于判断是否是重试消息）
     * @param message     消息内容（JSON 格式字符串）
     */
    @RabbitListener(queues = "your_queue_name") // 监听指定队列，自动消费消息
    public void processMessage(@Header("messageId") String messageId,
                               @Header(AmqpHeaders.REDELIVERED) Boolean redelivered,
                               String message) {
        try {
            log.info("Processing message: {}", message);
            JSONObject jsonObject = JSON.parseObject(message); // 解析 JSON 格式的消息
            handleMessage(jsonObject); // 处理业务逻辑
        } catch (Exception e) {
            log.error("Message processing failed: {}", message, e); // 记录错误日志

            // 查询数据库中是否已存在该失败消息的记录
            FailedMessageLog failedMessage = failedMessageLogMapper.findByMessageId(messageId);
            if (failedMessage == null) {
                // 如果数据库中没有记录，则插入一条新的失败记录
                failedMessage = new FailedMessageLog(messageId, message, e.getMessage(), 0, "PENDING");
                failedMessageLogMapper.insert(failedMessage);
            } else {
                // 如果已存在记录，则增加重试次数
                failedMessage.setRetryCount(failedMessage.getRetryCount() + 1);
                failedMessageLogMapper.update(failedMessage);
            }

            // 判断重试次数是否超过最大限制
            if (failedMessage.getRetryCount() >= MAX_RETRY_COUNT) {
                log.warn("Message {} reached max retry limit, moving to dead letter queue", messageId);
                failedMessage.setStatus("FAILED"); // 标记消息为失败
                failedMessageLogMapper.update(failedMessage); // 更新数据库记录

                // 将消息发送到死信队列（DLX），用于后续人工干预
                rabbitTemplate.convertAndSend("dlx_exchange", "dlx_routing_key", message);
            } else {
                log.warn("Retrying message: {}, attempt {}", messageId, failedMessage.getRetryCount());
                failedMessage.setStatus("RETRYING"); // 更新状态为重试中
                failedMessageLogMapper.update(failedMessage);

                // 发送到延迟队列，等待 5 秒后再重试，避免频繁失败造成系统压力
                rabbitTemplate.convertAndSend("delayed_exchange", "delayed_routing_key", message, msg -> {
                    msg.getMessageProperties().setDelay(5000); // 设置消息延迟 5 秒
                    return msg;
                });
            }
        }
    }

    /**
     * 业务处理逻辑（模拟成功处理）
     *
     * @param jsonObject 解析后的消息对象
     */
    private void handleMessage(JSONObject jsonObject) {
        log.info("业务处理成功: {}", jsonObject);
    }
}

📌 4. 配置延迟队列 & 死信队列

📍 延迟队列

@Bean
public Queue delayedQueue() {
    return QueueBuilder.durable("delayed_queue")
        .withArgument("x-dead-letter-exchange", "dlx_exchange")  
        .withArgument("x-dead-letter-routing-key", "dlx_routing_key")
        .withArgument("x-message-ttl", 10000)  // 10秒延迟
        .build();
}

📍 死信队列

@Bean
public Queue dlxQueue() {
    return QueueBuilder.durable("dlx_queue").build();
}

📌 5. 总结

这样，我们就可以 保证 RabbitMQ 消息消费的高可用性，同时避免消息丢失和死循环消费的问题。🚀🚀🚀