分布式系统中如何确保 MQ 消息不丢失、重复消费及积压处理

代码逻辑：

1. 尝试插入 msg_id 到 message_log 表。
2. 若插入成功，执行业务逻辑；若抛出唯一键冲突异常，说明已处理过，直接忽略。

2. Redis 原子操作

对于高性能场景，可使用 Redis 的 SETNX 命令实现分布式锁。

String key = "msg_lock:" + messageId;
Boolean locked = redisTemplate.opsForValue().setIfAbsent(key, "1", 5, TimeUnit.MINUTES);
if (locked) {
    try {
        // 执行业务逻辑
    } finally {
        redisTemplate.delete(key);
    }
} else {
    // 消息已被处理
}

3. 乐观锁版本号

在业务表中增加版本号字段。更新时检查版本号是否匹配，防止并发覆盖。

UPDATE account SET balance = balance - 100, version = version + 1 
WHERE id = 1 AND version = old_version;

三、消息积压的处理策略

消息积压通常发生在消费端处理能力不足时，表现为消费进度滞后。解决思路分为应急处理和根本优化。

1. 应急扩容

• 增加消费者实例：临时增加消费端机器数量，提高并发处理能力。
• 降级非核心业务：暂停非关键业务逻辑，集中资源处理核心消息。

2. 分区与并发优化

以 Kafka 为例，消费能力受限于 Topic 的 Partition 数量。一个 Partition 只能被一个消费者线程消费。

• 扩容原则：消费者实例数不应超过 Partition 总数。若消费者多于分区，多余实例将处于空闲状态。
• 动态调整：通过增加 Partition 数量来提升吞吐上限，随后增加消费者实例匹配新分区。

3. 批量消费与异步处理

• 批量拉取：配置消费者每次拉取多条消息（如 max.poll.records=500），减少网络交互开销。
• 异步处理：将消息放入内部线程池异步处理，快速释放 Broker 连接，但需注意事务边界控制。

四、分布式全局唯一 ID 生成

在实现幂等性和消息追踪时，需要全局唯一的消息 ID。常见的生成方案包括 UUID、数据库自增、Redis 号段和 Snowflake 算法。

1. Snowflake 算法详解

Snowflake 算法生成的 ID 为 64 位长整型，包含时间戳、机器 ID 和工作进程 ID。结构如下：

• 符号位（1 位）：固定为 0。
• 时间戳（41 位）：毫秒级，可用约 69 年。
• 数据中心 ID（5 位）：标识机房。
• 机器 ID（5 位）：标识服务器。
• 序列号（12 位）：同一毫秒内递增，最大 4096。

优点：趋势递增，利于数据库索引性能；无依赖外部服务；高性能。 缺点：依赖系统时钟，若时钟回拨可能导致 ID 重复。需实现时钟回拨补偿机制。

2. 选型建议

• 简单场景：UUID 足够，但无序，不适合做主键。
• 高并发主键：推荐 Snowflake 或其变种（如美团 Leaf）。需关注时钟同步问题。
• 强一致性：数据库自增或 Redis 号段，但存在单点瓶颈风险。

五、总结

确保 MQ 消息可靠性是一个系统工程，涉及生产、存储、消费全链路的设计。

1. 防丢失：生产端开启 Confirm/事务，Broker 开启持久化与多副本，消费端使用手动 ACK 与死信队列。
2. 防重复：构建幂等性机制，优先使用数据库唯一键或 Redis 锁。
3. 防积压：监控消费延迟，合理规划 Partition 数量，实施水平扩容与降级策略。
4. ID 生成：选择适合业务的高性能全局 ID 算法，如 Snowflake。

在实际落地中，需权衡性能与一致性。例如，强制同步刷盘会显著降低吞吐量，可根据业务容忍度选择异步刷盘配合定期备份。通过上述方案的组合应用，可构建高可用的消息驱动架构。