《Java 消息中间件·事件驱动架构》5.3 并发控制与版本

事件驱动中的并发写场景

CQRS 架构下，聚合根的状态变更由命令触发，事件写入事件日志。但如果两个消费者同时处理针对同一聚合的命令，可能并发修改同一笔贷款的状态：

• 消费者 A 处理 CancelLoan，读取聚合 v3，意图写 v4 LoanCancelled。
• 同时消费者 B 处理 ExtendDeadline，读取同一聚合 v3，意图写 v4 DeadlineExtended。

如果不加控制，两条 v4 事件会冲突：最终状态取决于谁后写入。这不是幂等问题，而是并发写冲突。

乐观并发控制的基本模型

乐观并发控制（Optimistic Concurrency Control, OCC）的核心假设是冲突不常发生。每次写操作携带「我基于的版本号」，存储端检查该版本是否仍是最新。如果是，允许写入，版本号加一；如果不是，拒绝并返回冲突。

以内存中的聚合状态为例：

import java.util.concurrent.ConcurrentHashMap;

class AggregateState {
    String status;
    int version;
    AggregateState(String status, int version) {
        this.status = status;
        this.version = version;
    }
}

class OptimisticStore {
    private final ConcurrentHashMap<String, AggregateState> store = new ConcurrentHashMap<>();

    boolean save(String id, int expectedVersion, String newStatus) {
        return store.compute(id, (k, current) -> {
            if (current == null || current.version == expectedVersion) {
                return new AggregateState(newStatus, expectedVersion + 1);
            }
            return current; // 版本不匹配，保留旧值
        }) != null && store.get(id).version == expectedVersion + 1;
    }

    AggregateState load(String id) { return store.get(id); }
}

模拟并发冲突实验

以下代码模拟两个线程同时竞争同一个聚合：

public class ConcurrencyDemo {
    public static void main(String[] args) throws InterruptedException {
        OptimisticStore store = new OptimisticStore();
        store.save("loan-99", 0, "INIT");

        Thread t1 = new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                AggregateState s = store.load("loan-99");
                boolean ok = store.save("loan-99", s.version, "CANCELLED");
                if (ok) { System.out.println("T1 success at v" + s.version); break; }
                System.out.println("T1 conflict at v" + s.version);
                try { Thread.sleep(10); } catch (InterruptedException e) {}
            }
        });

        Thread t2 = new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                AggregateState s = store.load("loan-99");
                boolean ok = store.save("loan-99", s.version, "EXTENDED");
                if (ok) { System.out.println("T2 success at v" + s.version); break; }
                System.out.println("T2 conflict at v" + s.version);
                try { Thread.sleep(10); } catch (InterruptedException e) {}
            }
        });

        t1.start(); t2.start(); t1.join(); t2.join();
        AggregateState f = store.load("loan-99");
        System.out.println("Final: " + f.status + " v" + f.version);
    }
}

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

javac --release 21 ConcurrencyDemo.java && java ConcurrencyDemo

输出示例：

T1 conflict at v1
T2 success at v1
T1 success at v2
Final: CANCELLED v3

两线程只有一个先成功，另一个检测版本变化后重试，最终以一致顺序完成更新。

冲突处理策略

检测到版本不匹配后，消费者有三种选择：

1. 重试：重新读取最新状态，重新应用业务逻辑，再次尝试写入。最常用但重试次数需上限，防止活锁。
2. 合并：两个变更不互斥时（如「增加标签」和「修改备注」），可合并为新事件。但合并逻辑与业务强耦合，难以通用。
3. 拒绝：直接异常，由人工或 Saga 补偿处理。适用于金额类、库存类等不能接受自动重试的场景。

推荐：策略 1 用于非关键字段更新，策略 3 用于金额/库存类变更。

重试退避策略

检测到冲突后盲目重试可能加剧竞争。合理退避：

• 固定间隔：每次冲突后等待固定毫秒。简单但多线程可能同时重试再冲突。
• 指数退避：10 → 20 → 40 → 80 毫秒，上限 200 毫秒。降低碰撞概率。
• 随机抖动：在固定间隔上加 0–50% 随机偏移，打散重试时间点。

Kafka 消费者的重试时间不能无限延长，因为 max.poll.interval.ms 限制两次 poll() 之间的最大间隔。如果重试超过该阈值，消费者会被踢出 consumer group。

与 Kafka 消费语义的关系

消费者线程池模型下，同聚合的消息可能被不同线程并行处理。即使 Kafka 保证分区有序，消费端仍会出现并发写。

解决方案有二：

• 按聚合 ID 路由到固定线程：消费者内部维护线程池，用 aggregateId.hashCode() % poolSize 决定处理线程。同一聚合的所有事件串行处理，从根本上避免并发写。
• 聚合内用乐观锁、聚合间并行：允许不同聚合的事件并行处理，同一聚合通过 OCC 串行化。这是 CQRS 的典型做法。

悲观锁 vs 乐观锁

事件驱动微服务中，通常单一聚合的并发写频率不高，乐观锁更符合常规负载。

投影层的并发写

CQRS 的投影也可能出现并发写。当两个消费者分别处理同一聚合的不同事件时，都要更新投影表：

UPDATE loan_read_model
SET status = ?, version = ?
WHERE aggregate_id = ? AND version = ?

WHERE version = ? 是乐观锁的体现。如果返回 updatedRows == 0，说明其他消费者已写入新版本，当前事件应重试或跳过。

持久化版本号的原子性

版本号的「读取-校验-写入」必须是原子的。常见实现：

最简方案是 H2 内存数据库：UPDATE 的原子性由数据库保证。

冲突与幂等的区分

冲突（concurrency conflict）和重复（duplicate）是两个独立维度：

消费者应先做幂等检查，再做乐观并发写入。去重失败跳过不触发 OCC；OCC 失败的进入重试或拒绝逻辑。

并发冲突的数量级估算

假设系统有 100 万活跃聚合，每秒产生 1000 个事件。事件均匀分布时，同一聚合的并发写概率极低（λ≈0.001）。如果存在热点聚合（如爆款商品的库存），λ 可能上升到 10 以上。

热点聚合的缓解方案：

• 拆分聚合：按仓库维度拆分为多个子聚合。
• 引入缓冲层：用本地队列串行化热点聚合的写请求。
• 降级为悲观锁：对热点聚合使用数据库行锁，强制串行化。

OCC 与生产者事务的联动

Kafka producer 的幂等发送（enable.idempotence）保证 broker 端不重复写入，但不保证消费端不重复处理。OCC 在消费端提供了「写前检查」机制，与生产者事务形成互补：

• 生产者端：PID + Sequence Number 防止 broker 重复落盘。
• 消费者端：版本号 CAS 防止重复处理触发重复副作用。

如果消费者使用事务性消费（Kafka Transactions），consume-transform-produce 循环中可以把「OCC 检查 → 投影写入 → offset 提交」放在一个事务里。但本卷限定不使用 Spring Kafka，纯 kafka-clients 的事务 API 较为底层，配置复杂。对于绝大多数场景，手动提交 offset + 数据库层面的 OCC 已足够可靠。

实际场景：libraryd 的并发修改

libraryd 的贷款状态可能被两个独立操作同时修改：

1. 用户点击「延期还款」，系统读取 v3，意图写入 v4 DeadlineExtended。
2. 风控系统检测到逾期，自动触发逾期标记，同样读取 v3，意图写入 v4 LoanOverdue。

这两个操作没有直接的调用关系，但都命中同一个聚合根。如果没有 OCC，最终结果取决于网络延迟和数据库写入顺序——用户可能刚申请延期就收到逾期通知，或反之。

OCC 的介入方式是：两个操作同时尝试 CAS 写入 v4，只有一个成功。失败者根据业务规则选择重试（如延期申请重新读取最新版本，发现已逾期则拒绝延期）或拒绝（如风控系统发现贷款已延期，跳过逾期标记）。

这个例子说明，OCC 不仅解决并发写的数据一致性问题，还通过「失败 → 重试/拒绝」路径暴露业务流程中的时序冲突，帮助发现不合理的并发设计。

冲突率与吞吐量的权衡

OCC 在高冲突场景下会成为瓶颈。假设 1000 个并发消费者同时竞争 10 个热点聚合，冲突率可能超过 50%。每次冲突触发重试，重试又消耗 CPU 和内存，形成恶性循环。

缓解方案：

1. 聚合拆分：把一个大聚合拆成多个子聚合。例如把「订单」按商品拆分为「订单-商品-A」「订单-商品-B」，每个子聚合独立竞争，冲突率随子聚合数量线性下降。
2. 引入写缓冲：对热点聚合使用内存队列串行化写请求。所有对同一聚合的写操作先进入队列，单线程消费队列执行 CAS，消除并发冲突。
3. 降级为串行分区：对热点聚合使用 Kafka 的 partition key 强制进入单一分区，消费者单线程串行消费，从源头消除并发。

这三种方案的取舍取决于系统的读写比例和延迟要求。

冲突的日志与监控

频繁冲突意味着聚合粒度过大或消费者并行度过高。建议记录：

CONFLICT aggregate=loan-99 expected=3 current=5 event=LoanCancelled handler=consumer-2

汇总 CONFLICT 日志可形成冲突热力图，指导聚合拆分或线程池调整。

本节小结

• 事件驱动系统中，同一聚合的并发写需要显式控制。
• 乐观并发控制用版本号检测冲突，不阻塞其他线程，适合冲突率低的场景。
• 冲突处理策略：重试（通用）、合并（业务耦合）、拒绝（关键字段）。
• 退避策略必须考虑 max.poll.interval.ms 限制，避免消费者被踢出组。
• 消费者内部按 aggregateId 路由到固定线程，可从根本上避免聚合内并发写。
• 投影层用 UPDATE ... WHERE version = ? 实现原子条件更新。
• 冲突日志是发现聚合粒度过大的有效工具。
• OCC 与生产者幂等形成互补，覆盖 broker 到消费端的完整一致性链路。
• ConcurrentHashMap.compute() 的 CAS 语义由 JDK Unsafe 实现，无需额外显式锁。
• H2 的 UPDATE ... WHERE version = ? 利用数据库行锁实现原子条件更新，无需应用层锁。
• 冲突日志格式建议统一为 CONFLICT aggregate=... expected=... current=... handler=... evt=...。
• 热点聚合的缓解方案包括聚合拆分、写缓冲和降级为串行分区三种策略。
• 聚合拆分按业务维度（如仓库、地区）将大聚合拆为多个子聚合。
• 写缓冲用本地队列串行化热点聚合的写请求，消除并发冲突。
• 降级为串行分区通过 Kafka partition key 强制进入单一分区实现，由单线程消费。

阅读导航：上一节 5.2 幂等边界的设计 ｜ 下一节 6.1 事件版本化

• 并发控制策略的选择应基于实际业务场景的冲突率和延迟要求。