Java 消息可靠性投递机制与方案

3.3 消费端保证

// 消费端保证示例
@RabbitListener(queues = "order.queue")
public void handleOrder(OrderMessage order, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
    try {
        // 1. 业务处理
        orderService.process(order);
        // 2. 手动确认（成功才 ACK）
        channel.basicAck(tag, false);
        // 3. 更新消费记录
        consumeRecordService.markConsumed(order.getId());
    } catch (Exception e) {
        // 4. 失败处理：重试或进入死信队列
        if (retryCount < MAX_RETRY) {
            channel.basicNack(tag, false, true); // 重入队列
        } else {
            channel.basicNack(tag, false, false); // 进入死信队列
            alarmService.notifyAdmin(order, e);
        }
    }
}

消费端关键点：

• 手动 ACK：避免自动确认导致消息丢失
• 幂等性设计：

public boolean processWithIdempotent(String msgId) {
    // 基于消息 ID 去重
    if (redis.exists("processed:" + msgId)) {
        return true; // 已处理过
    }
    // 业务处理
    boolean success = doBusinessLogic();
    // 记录处理状态
    if (success) {
        redis.setex("processed:" + msgId, 24h, "1");
    }
    return success;
}

• 死信队列（DLQ）：处理无法消费的消息
• 消费重试策略：
  • 立即重试（瞬时故障）
  • 延迟重试（业务依赖）
  • 指数退避（防止雪崩）

4. 完整可靠性方案

4.1 事务消息方案（如 RocketMQ）

两阶段提交：
1. 发送 Half Message（预备消息）
2. 执行本地事务
3. 根据本地事务结果 Commit/Rollback
4. Broker 检查事务状态并投递/丢弃

4.2 最大努力投递方案

# 补偿机制实现
def reliable_delivery(message):
    max_retries = 5
    for attempt in range(max_retries):
        try:
            # 尝试投递
            result = mq_client.send(message)
            if result.confirmed:
                log_delivery_success(message.id)
                return True
        except Exception as e:
            log_failure(attempt, e)
            if attempt == max_retries - 1:
                # 最终失败，人工介入
                send_alert_to_admin(message)
                save_to_compensation_table(message)
                return False
        # 等待后重试
        sleep(backoff_time(attempt))
    return False

4.3 本地消息表方案（经典）

-- 本地消息表结构
CREATE TABLE local_message (
    id BIGINT PRIMARY KEY,
    biz_id VARCHAR(64), -- 业务 ID
    content TEXT, -- 消息内容
    status TINYINT, -- 0:待发送，1:已发送，2:已确认
    retry_count INT,
    next_retry_time DATETIME,
    created_at TIMESTAMP
);

工作流程：

1. 业务数据 + 消息记录原子性写入本地 DB
2. 定时任务扫描待发送消息
3. 调用 MQ 发送，成功后更新状态
4. 消费者处理完成后反向确认
5. 对账程序定期校验数据一致性

5. 高级特性与优化

5.1 顺序性保证

• 全局有序：单队列单消费者（性能低）
• 局部有序：相同 sharding key 的消息发到同一队列
• 牺牲场景：重试队列可能破坏顺序

5.2 批量消息可靠性

// 批量消息的可靠性处理
public class BatchMessageReliableSender {
    public void sendBatch(List<Message> batch) {
        // 1. 批量持久化到本地
        batchMessageDao.saveAll(batch);
        // 2. 设置批次 ID
        String batchId = generateBatchId();
        // 3. 发送批次消息
        boolean success = mqTemplate.sendBatch(batchId, batch);
        // 4. 批次确认（或单条补偿）
        if (success) {
            markBatchDelivered(batchId);
        } else {
            // 逐条重试或记录异常
            compensateFailedMessages(batch);
        }
    }
}

5.3 监控与对账

• 实时监控：
  • 堆积情况监控
  • 消费延迟报警
  • 失败率统计
• 定期对账：

-- 消息对账 SQL 示例
SELECT DATE(create_time) as day, COUNT(*) as total_sent,
       SUM(CASE WHEN status=2 THEN 1 ELSE 0 END) as confirmed,
       SUM(CASE WHEN status=1 THEN 1 ELSE 0 END) as pending
FROM message_record
GROUP BY DATE(create_time)
HAVING total_sent != confirmed;

6. 不同 MQ 的实现差异

7. 实践建议

1. 分级可靠性策略：
   • 关键业务：事务消息 + 本地表 + 对账
   • 普通业务：确认机制 + 重试 + 死信队列
   • 日志类：最多一次投递即可
2. 性能与可靠性的平衡：
   • 同步刷盘 vs 异步刷盘
   • 同步复制 vs 异步复制
   • 根据业务重要性选择配置
3. 灾难恢复设计：

# 配置示例：多级降级
mq:
  primary:
    url: "amqp://primary"
    timeout: 1000ms
  secondary:
    url: "amqp://secondary"
    timeout: 2000ms
  fallback-to-db: true # 最终降级到数据库

总结

消息的可靠性投递是一个系统工程，需要在生产端、Broker 端、消费端协同设计，结合业务场景、性能要求、成本约束做出合适的选择。没有"银弹"方案，只有最适合的方案。建议从简单方案开始，随着业务复杂度增加逐步引入更完善的可靠性机制。

面试回答

首先，消息可靠性投递指的是：
一个消息从发送到被消费者成功处理，过程中不会丢失或重复，保证最终数据的一致性。在实际系统里，消息可能因为网络问题、服务重启等原因丢失或重复，所以我们需要一套机制来确保可靠。

为什么需要它呢？
比如在订单系统中，用户支付成功后要通知物流系统，如果消息丢了，物流就不会触发，用户体验就受损；如果消息重复，可能重复发货，造成损失。所以像金融、交易这些场景，可靠性特别重要。

常见的实现方式，我了解的有几种：

1. 生产者确认机制
   生产者发消息后，MQ（比如 RabbitMQ）会返回一个确认（ACK），如果没收到 ACK，生产者可以重发。这样可以防止消息在发送阶段丢失。
2. 消息持久化
   消息保存到磁盘，而不是只放在内存。这样即使 MQ 重启，消息也不会丢。
3. 消费者手动 ACK
   消费者处理完消息后，手动告诉 MQ'我已经处理完了'，MQ 才删除消息；如果处理失败，MQ 可以把消息重新投递给其他消费者。避免消息在处理阶段丢失。
4. 事务消息（比如 RocketMQ）
   先发一个'半消息'，等本地事务执行成功，再确认投递；如果失败，就回滚。这适用于分布式事务场景。
5. 消息去重
   为了避免重复消费，可以在消费端做幂等性设计。比如在数据库里记录消息 ID，每次处理前先查一下是否已经处理过。

实际中我们一般会结合业务来设计。
比如一个订单状态同步的场景，我可能会用：生产者确认 + 消息持久化 + 消费者手动 ACK + 消费端幂等性。这样基本能覆盖发送、存储、消费各个环节的可靠性。

当然，可靠性和性能之间需要权衡，比如持久化会降低吞吐量，手动 ACK 会增加延迟。所以要根据业务需求来选择合适的方案。

追加：遇到过消息丢失或重复的问题，你是怎么排查和解决的？

追加：是否了解最终一致性、最大努力通知等模式？