如何保证消息一致性

消息一致性保障：从发送到消费的全链路解决方案

消息一致性是分布式系统中消息队列使用的核心挑战，指业务操作与消息传递的状态保持一致：业务成功时消息必须正确发送并被处理，业务失败时消息不应被发送或需被撤回。本文从消息发送一致性和消息消费一致性两方面，解析问题根源与解决方案。

消息发送一致性：确保业务与消息的原子性

消息发送一致性的核心目标是：业务操作成功 ↔ 消息必须发送成功，避免 “业务成功但消息丢失” 或 “业务失败但消息发送” 的矛盾场景。

消息发送可能出现的问题

（1）业务成功但消息未发送

场景 1：业务逻辑执行完成后，应用在发送消息前宕机（如 JVM 崩溃），导致消息未发送。
场景 2：业务成功后，消息发送时消息中间件宕机（如 Broker 崩溃），消息未被持久化。

（2）业务失败但消息已发送

场景：消息先发送成功，但后续业务逻辑执行失败（如数据库事务回滚），导致无效消息被消费。

解决方案：基于 “消息预存 + 状态确认” 的两阶段方案

用户提到的方案本质是 “预发送消息→执行业务→确认消息可用” 的两阶段模式，确保业务与消息状态同步。具体流程如下：

步骤拆解：

预存消息：业务应用先向消息中间件发送 “待确认” 消息（状态标记为未处理），消息中间件仅存储消息但不投递。
- 目的：先确保消息能被中间件持久化，避免后续业务成功后消息发送失败。
消息中间件反馈存储结果：中间件返回 “消息存储成功 / 失败”。
- 若失败：业务直接终止（避免业务成功但消息无法存储）。
执行业务逻辑：仅当消息存储成功后，才执行核心业务（如订单创建、库存扣减）。
确认消息状态：业务执行完成后，向中间件发送 “业务结果”：
- 业务成功：中间件将消息状态改为可投递，并向消费者投递。
- 业务失败：中间件删除预存消息（或标记为废弃），避免无效消息。

关键保障：

原子性：消息预存是业务执行的前提，业务结果直接决定消息是否生效，二者强绑定。
持久化：预存消息需持久化（如写入磁盘），避免中间件宕机导致消息丢失。

业界主流实现：本地消息表与事务消息

上述方案可通过两种方式落地，解决分布式环境下的原子性问题：

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

消息一致性

原理：在业务数据库中创建 “消息表”，将消息发送与业务操作放入同一本地事务。
流程：
1. 业务开始时，在本地事务中插入一条 “待发送” 消息到消息表（与业务数据同库，确保原子性）。
2. 事务提交后，通过定时任务（或消息队列）扫描 “待发送” 消息，发送到中间件。
3. 消息发送成功后，更新消息表状态为 “已发送”；失败则重试。
优势：不依赖中间件特性，任何消息队列均可使用。
示例：订单系统在orders表插入订单的同时，在order_messages表插入一条 “订单创建” 消息，确保两者同时成功或失败。

（2）事务消息（中间件原生支持）

原理：消息中间件提供事务消息机制（如 RocketMQ 的 TransactionMQ），通过 “半消息 + 确认” 机制实现原子性。
流程：
1. 生产者发送 “半消息”（Half Message）到中间件，中间件存储但不投递（类似 “预存消息”）。
2. 中间件回调生产者的checkLocalTransaction方法，查询业务执行结果。
3. 若业务成功，生产者返回COMMIT，中间件投递消息；若失败，返回ROLLBACK，中间件删除消息。
优势：无需维护本地消息表，由中间件保证一致性，简化开发。

消息消费一致性：确保消息被正确处理

消息消费一致性的核心目标是：消息必须被正确处理且仅处理一次，避免 “消息丢失”“重复消费” 或 “处理失败但被标记为成功”。

消费端可能出现的问题

（1）消息丢失

场景：消费者接收到消息，但处理过程中宕机，且未向中间件发送确认，导致消息被重新投递时丢失上下文（如处理到一半的业务）。

（2）重复消费

场景：消费者处理完成后，向中间件发送确认时网络中断，中间件未收到确认，会重新投递消息，导致重复处理（如重复扣减库存）。

（3）处理失败但消息被确认

场景：消费者错误地在处理前发送确认，后续处理失败，消息无法重新投递。

解决方案：基于 “确认机制 + 幂等处理” 的可靠消费

（1）手动确认机制（避免消息丢失）

关闭消费者的 “自动确认”（autoAck=false），改为处理完成后手动确认：
1. 消费者接收消息后，先不确认，执行业务逻辑（如数据库操作）。
2. 业务处理成功后，调用basicAck（RabbitMQ）或commitOffset（Kafka）发送确认。
3. 若处理失败，调用basicNack（RabbitMQ）拒绝消息，中间件会重新投递（或进入死信队列）。

示例（RabbitMQ）：

channel.basicConsume(queueName, false, (consumerTag, delivery) -> {
    // 1. 处理业务
    boolean success = processBusiness(delivery.getBody());
    // 2. 手动确认
    if (success) {
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } else {
        // 失败则重新投递（最多3次）
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
    }
}, consumerTag -> {});

（2）幂等处理（避免重复消费）

无论中间件是否重复投递，消费者需保证重复消息处理结果一致，核心是 “去重”：
- 方案 1：唯一 ID + 幂等表：消息携带唯一 ID（如订单 ID），处理前先查询幂等表（如message_processed），若已处理则直接返回成功。
- 方案 2：业务天然幂等：设计业务操作时保证幂等（如UPDATE inventory SET num = num - 1 WHERE id = 1 AND num > 0，重复执行结果一致）。

（3）死信队列与重试机制（处理失败消息）

对多次处理失败的消息（如因依赖服务宕机），不应无限重试，需进入死信队列（Dead Letter Queue）：
1. 配置重试次数上限（如 3 次），超过后消息进入死信队列。
2. 通过人工干预或定时任务处理死信队列中的消息（如排查依赖服务问题后重新投递）。

最终一致性保障：异常补偿机制

分布式系统中无法完全避免网络分区、服务宕机等异常，需通过补偿机制确保 “最终一致性”：

定时任务校验：
- 发送端：定期扫描本地消息表中 “待发送” 超过阈值的消息，重新发送（解决 “业务成功但消息未确认” 问题）。
- 消费端：定期比对 “已接收消息” 与 “业务处理记录”，发现未处理的消息触发重试。
消息轨迹追踪：
- 为每条消息添加唯一 ID，记录发送、投递、消费的全链路状态（如通过中间件的消息追踪功能），便于排查异常。
监控告警：
- 监控消息队列的 “未确认消息数”“死信队列长度”“消费延迟” 等指标，超过阈值时告警，及时介入处理