《Java 消息中间件·事件驱动架构》5.2 幂等边界的设计

“幂等消费者"是一个边界声明

Kafka consumer 配置里都没有 enable.idempotence 这个开关。生产者有 ENABLE_IDEMPOTENCE_CONFIG（默认 true），它保证在 broker 端不重复写入相同消息。但消费者的重复处理完全留给业务代码。

因此在事件驱动架构中，幂等不是全局属性，而是消费者与状态存储之间的边界函数。具体来说，就是「用什么样的键去重」这个设计决策。

三种幂等键的对比

消息 ID 是最精确的方案，前提是生产者给每条消息分配全局唯一 ID。业务键覆盖的是业务实体的生命周期，适合 CQRS 投影场景。自然键最轻量，但依赖操作语义，如「状态设为 overdue」无论执行多少次结果相同，而「余额增加 100」就不是自然幂等的。

用 messageId 去重的内存实现

以下代码演示了基于消息 ID 的消费者端去重：

import java.util.*;

class MessageIdDeduplicator {
    private final Set<String> seen = new HashSet<>();
    private int capacity;

    MessageIdDeduplicator(int capacity) { this.capacity = capacity; }

    boolean isDuplicate(String messageId) {
        if (messageId == null || messageId.isBlank()) return false;
        return !seen.add(messageId);
    }
}

局限性：只在进程内存中生效。当消费者发生重平衡、分区迁移到另一台实例时，新实例没有旧的 seen 集合。除非把去重状态写入 Redis 或数据库，否则消息 ID 去重只能作为短窗口内的防护，不能作为跨实例的强保证。

用业务键去重：最实用的中长期方案

在 CQRS 投影中，消费者把事件写入一张投影表。表的主键就是 (aggregate_id, version)。同一聚合的同一版本事件无论收到多少次，主键冲突只允许写入一次。

消费者收到事件后，用 INSERT 而不是 SELECT → UPDATE。DuplicateKeyException 即表示幂等命中，跳过即可：

void upsert(String aggregateId, int version, String type, double balance) {
    String sql = "INSERT INTO loan_projection VALUES (?, ?, ?, ?)";
    try (PreparedStatement ps = conn.prepareStatement(sql)) {
        ps.setString(1, aggregateId);
        ps.setInt(2, version);
        ps.setString(3, type);
        ps.setDouble(4, balance);
        ps.executeUpdate();
    } catch (SQLException e) {
        if (e.getErrorCode() == 23505) { // H2 duplicate key
            System.out.println("DUP " + aggregateId + " v" + version);
        } else {
            throw new RuntimeException(e);
        }
    }
}

这种方法的幂等边界是数据库主键约束。只要 (aggregate_id, version) 在业务层面唯一，消费者重启、重平衡、代码 bug 导致的重复 poll 都能被拦截。

自然键幂等：零存储成本

如果事件本身的语义具有幂等性，就不需要任何外部存储：

• LoanOverdue：把贷款状态设为 OVERDUE。执行两次结果一样。
• LoanActivated：把状态设为 ACTIVE，同样幂等。
• BalanceIncremented(amount=100)：不是自然幂等，第二次执行会让余额多 100。

自然键幂等的判定标准是「操作的最终状态不依赖于执行次数」。event sourcing 中，如果只维护最终状态快照（如 current_status），那么「设值」类事件就是自然幂等的。但如果同时维护审计日志或余额累加，就需要额外防护。

跨聚合的幂等边界

单一消费者的去重无法覆盖「多条消息触发同一副作用」的场景。例如：

• 订单服务发出 OrderCreated(order-100)
• 库存服务处理 OrderCreated 并发出 StockReserved(goods-50)
• 支付服务订阅 StockReserved，执行扣款

如果支付服务因网络超时重新消费 StockReserved，它应在支付记录表中检查 (orderId, eventVersion) 而不是只检查消息 ID。因为消息 ID 去重只保证单条消息不重复处理，但支付服务的副作用（创建支付记录）是跨聚合的。

跨聚合幂等的关键是副作用的事务性：去重键必须覆盖副作用涉及的所有业务实体。

幂等与「至少一次」语义

Kafka 消费者默认 enable.auto.commit=true，每隔一段时间自动提交 offset。但如果把 enable.auto.commit=false 改成手动提交，流程变成：

1. poll 一批消息
2. 逐条写入投影（含去重键）
3. commitSync() 提交 offset

此时如果步骤 2 成功、步骤 3 失败（进程崩溃），重启后这批消息会重发。去重键（如数据库主键）在此时起决定性作用，防止重复写入投影。这比「exactly once」在 broker 层面的语义更可靠，因为去重发生在业务存储层。

可运行的幂等判定实验

以下代码把三种方案放在同一判定流程中，模拟消费者收到重复消息的场景：

import java.util.*;

class IdempotencyJudge {
    Set<String> messageIds = new HashSet<>();
    Map<String, Integer> aggregateVersions = new HashMap<>();

    String judge(String messageId, String aggregateId, int version,
                 String eventType) {
        if (messageId != null && messageIds.contains(messageId)) {
            return "MESSAGE_ID_DUP";
        }
        Integer cur = aggregateVersions.get(aggregateId);
        if (cur != null && version <= cur) {
            return "VERSION_DUP";
        }
        if (eventType.equals("LoanOverdue") || eventType.equals("LoanActivated")) {
            return "NATURAL_IDEMPOTENT";
        }
        messageIds.add(messageId);
        aggregateVersions.put(aggregateId, version);
        return "APPLIED";
    }
}

测试输入：

public class IdempotencyDemo {
    public static void main(String[] args) {
        IdempotencyJudge judge = new IdempotencyJudge();
        String id = "msg-uuid-42";
        System.out.println(judge.judge(id, "loan-1", 1, "LoanCreated"));
        System.out.println(judge.judge(id, "loan-1", 1, "LoanCreated"));
        System.out.println(judge.judge(id + ":alt", "loan-1", 0, "LoanOverdue"));
    }
}

编译运行：

javac --release 21 IdempotencyDemo.java && java IdempotencyDemo

输出：

APPLIED
MESSAGE_ID_DUP
NATURAL_IDEMPOTENT

（单次运行，随调度波动，不作基准）

幂等键的选择决策树

1. 消息体里有 messageId → 用消息 ID 去重，短期有效，长期需要持久化到外部存储。
2. 消息体里没有 messageId，但有 aggregateId 和 version → 用数据库主键约束去重，最可靠。
3. 事件语义天然幂等（设值类）→ 零额外成本，每次直接写入即可。
4. 以上都不满足 → 在投影层补充 (targetEntity, expectedVersion) 字段，通过乐观锁实现去重。

幂等键的过期策略

长期运行的系统中，去重表会不断膨胀。业务键去重可以按业务生命周期清理：贷款结清后，其所有历史版本的去重记录可以归档删除。消息 ID 去重则需要设置 TTL——内存中是 LRU，外部缓存中设过期时间。

一个简单做法是在去重表中增加 last_seen_at 字段，定时删除 30 天前的记录。过期策略的风险在于 TTL 窗口内发生重平衡：如果一条消息在 29 天前被处理，其去重记录在第 30 天过期，但恰好此时消费者重平衡导致消息重发，去重不再命中。

生产者幂等与消费者幂等的边界

Kafka producer 的 enable.idempotence=true 保证的是 broker 接收端不重复写入。如果 producer 重试发送同一条消息，broker 通过 PID + Sequence Number 识别并去重。但这只解决「写入不重复」，不解决「消费不重复」。

因此，完整的幂等链路需要两端配合：

• 生产者：开启幂等发送，确保 broker 不重复落盘。
• 消费者：应用层去重，确保重复投递不触发重复副作用。

如果仅开启生产者幂等、不实现消费者去重，在 consumer 重平衡后仍会出现重复处理。这也是 event-driven 架构中「至少一次」语义长期存在的根本原因。

幂等与顺序的关系

幂等和顺序是独立的两个维度：

「非幂等 + 乱序」是最少见也最危险的组合。它通常出现在金融交易类系统中，如「先扣款再发货」不能变成「先发货再扣款」，且扣款不能重复执行。这类场景需要完整的 Saga 编排，不能仅凭消费者去重。

幂等键容量的估算

业务键去重表的行数等于系统中活跃聚合的事件总量。假设每个贷款聚合平均产生 50 个事件，100 万笔贷款的去重表行数为 5000 万。按每行 32 字节估算（aggregate_id 16 字节 + version 4 字节 + 索引开销），总存储约 1.6 GB。

消息 ID 去重的内存占用：每条消息 ID 按 36 字节 UUID 计算，10 万条消息的去重集约 3.6 MB。如果使用 LRU 策略限制在 1 万条，内存降至 360 KB，代价是 TTL 外的重发无法命中。

这个估算说明，业务键去重的存储成本在可接受范围内，且天然持久化；消息 ID 去重虽然内存轻量，但持久化后成本与业务键方案趋同。

去重键与事务一致性

去重键必须和副作用在同一事务内提交，否则会出现「去重已记录但副作用未执行」的中间态。以 H2 为例，消费者应在同一个 Connection 中：

conn.setAutoCommit(false);
try {
    upsertDedup(aggregateId, version);    // 去重表 INSERT
    updateProjection(aggregateId, status); // 投影表 UPDATE
    conn.commit();
} catch (SQLException e) {
    conn.rollback();
    throw e;
}

如果 upsertDedup 成功、updateProjection 失败，回滚会同时撤销两者。否则，如果 upsertDedup 已提交但投影未更新，重平衡恢复后同一事件会被判定为已处理而跳过，但投影数据缺失，形成数据不一致。

事务范围的另一个约束是 Kafka offset 提交。commitSync() 应在数据库事务提交之后调用，这样即使 offset 提交失败，最多只是重复消费，不会丢失数据。

去重方案的演进路径

对于大多数系统，推荐按以下路径演进：先从自然键幂等开始，业务复杂化后引入业务键去重，最后在高并发场景下补充消息 ID 去重。这个顺序从简单到复杂，每步都有明确的触发条件和回退能力。

自然键适用于状态机类场景，业务键是 CQRS 投影的标配，消息 ID 去重则用于跨消费者实例的强一致性保障。

去重失败的排查清单

如果观察到重复副作用，按以下顺序排查：

1. 检查消费者是否跨实例：去重状态存储在内存中的话，消费者重平衡后新实例没有历史记录。解决：持久化到数据库。
2. 检查去重表是否被清理：TTL 或归档任务删除了活跃聚合的去重记录。解决：按业务生命周期而非固定时间清理。
3. 检查事务边界：去重记录和副作用不在同一事务内提交。解决：使用同一个 Connection 的 setAutoCommit(false)。
4. 检查消息 ID 来源：生产者是否给每条消息分配了唯一 ID。解决：在发送前用 UUID.randomUUID() 生成。
5. 检查版本号递增：聚合根的版本号是否严格递增，有无并发写导致同版本号。解决：5.3 的乐观并发控制。

本节小结

• 消费者的幂等去重在应用层实现，Kafka consumer 没有内置机制。
• 消息 ID 去重精确，但受限于实例内存；持久化后又变成数据库主键去重的另一种形式。
• 业务键去重（aggregateId + version）是最实用的中期方案，利用数据库主键约束天然防重。
• 自然键去零零存储开销，但只适用于「设值」类语义。
• 跨聚合副作用的去重键必须覆盖所有涉及的业务实体。
• 手动提交 offset + 数据库主键去重，比自动提交更可靠地防止重复副作用。
• 生产者幂等只保证写入不重复，完整链路仍需消费者自己去重。
• 去重表存储成本可按活跃聚合事件量估算，常规场景下 1–2 GB 内可控。
• 去重键的选择必须与数据库事务范围一致，否则可能出现去重命中但副作用未提交的中间态。

阅读导航：上一节 5.1 事件顺序与业务顺序 ｜ 下一节 5.3 并发控制与版本