网络分区 → 应答超时 → 生产者重试 → 消息重复
消费者超时 → Broker重投 → 消息重复
消费者崩溃 → 位移未提交 → 重启后重复消费

-- 去重表设计
CREATE TABLE message_idempotent (
    message_id VARCHAR(128) PRIMARY KEY,  -- 消息唯一标识
    business_id VARCHAR(64),              -- 业务ID
    status TINYINT DEFAULT 0,             -- 处理状态：0-未处理，1-已处理
    processed_time DATETIME,              -- 处理时间
    created_time DATETIME DEFAULT NOW()   -- 创建时间
) ENGINE=InnoDB;

-- 处理流程伪代码
BEGIN TRANSACTION;
-- 1. 插入去重记录（唯一索引保证原子性）
INSERT INTO message_idempotent(message_id, business_id)
VALUES ('msg_001', 'order_123')
ON DUPLICATE KEY UPDATE status=status;

-- 2. 查询是否已处理
SELECT status FROM message_idempotent WHERE message_id='msg_001';

-- 3. 根据状态执行业务逻辑
IF status=0 THEN
    -- 执行业务操作
    UPDATE orders SET status='paid' WHERE order_id='order_123';
    -- 更新处理状态
    UPDATE message_idempotent SET status=1, processed_time=NOW()
    WHERE message_id='msg_001';
END IF;
COMMIT;

订单状态机示例：
[待支付] → (支付消息) → [已支付] → (发货消息) → [已发货]

幂等逻辑：
1. 收到支付消息时，检查当前状态是否为"待支付"
2. 只有状态为"待支付"时才执行扣款、更新状态操作
3. 如果状态已是"已支付"，直接返回成功，不做任何操作

// 基于数据版本号的幂等实现
public boolean processWithVersion(Message msg) {
    // 从消息中获取业务数据的当前版本
    int expectedVersion = msg.getDataVersion();

    // 使用CAS方式更新
    String sql = "UPDATE business_table " +
                 "SET status = ?, version = version + 1 " +
                 "WHERE id = ? AND version = ?";

    int rows = jdbcTemplate.update(sql,
        msg.getNewStatus(),
        msg.getBusinessId(),
        expectedVersion);

    // rows>0表示更新成功，说明版本匹配且是第一次处理
    // rows=0表示版本不匹配，说明已被其他请求处理过
    return rows > 0;
}

消息ID = 业务前缀 + 时间戳 + 机器ID + 序列号
示例：ORDER_20240115143000_NODE01_0001

设计要点：
1. 时间戳：精确到毫秒，保证时间有序
2. 机器ID：防止多节点冲突
3. 序列号：同一毫秒内的自增序列
4. 业务前缀：便于问题排查和分类处理

public boolean distributedIdempotentProcess(String messageId) {
    // 1. 获取分布式锁（基于消息ID）
    String lockKey = "lock:" + messageId;
    boolean locked = redis.setnx(lockKey, "1", 30);

    if (!locked) {
        // 锁获取失败，说明正在被其他节点处理
        return checkProcessStatus(messageId);
    }

    try {
        // 2. 检查是否已处理（临界区操作）
        if (isProcessed(messageId)) {
            return true;
        }

        // 3. 执行业务逻辑
        doBusiness();

        // 4. 标记为已处理
        markAsProcessed(messageId);

        return true;
    } finally {
        // 5. 释放锁
        redis.del(lockKey);
    }
}