codesphere

外观

DDK Concurrency Starter ​

按领域概念做并发控制：同一个聚合实例同时只有一个操作在处理，同一个客户端请求只执行一次，每个用户或租户有各自的调用额度。底层是 Redisson，本 starter 只补「按聚合加锁」「锁包住事务」「按请求去重」「按业务维度限流」这几个约定。

text
@RateLimit           为这个 key 取一个许可                   没有了 → RateLimitedException（429）
  @Idempotent          登记请求（SET NX + 过期时间）          重复 → DuplicateRequestException（409）
    @AggregateLock       获取 ddk:lock:<类型>:<标识>           等待时间内拿不到 → AggregateBusyException（409）
      @Transactional       加载 → 修改 → 保存 → 提交
    释放锁                                                  在提交之后，下一个操作加载到的是已提交的状态
失败时：释放锁，撤销请求登记

能做什么 ​

能力说明
@AggregateLock方法执行期间独占一个聚合实例，锁在事务之外获取、事务结束之后释放
@Idempotent同一个请求 key 在有效期内只执行一次，执行失败时可以用同一个 key 重试
@RateLimit每个 key（用户、租户）在一个时间窗口内最多执行若干次，所有实例共享计数
AggregateLocks编程式入口，用于不是 Spring Bean 方法的地方
错误码AGGREGATE_BUSY、DUPLICATE_REQUEST 映射为 409，RATE_LIMITED 映射为 429

引入方式 ​

xml
<dependency>
    <groupId>com.domaindrivenkit</groupId>
    <artifactId>ddk-concurrency-starter</artifactId>
</dependency>
java
@Transactional
@AggregateLock(type = "order", id = "#command.orderId()")
@Idempotent(key = "#command.requestId()")
public OrderResponse pay(PayOrderCommand command) {
    Order order = orders.findById(command.orderId()).orElseThrow(...);
    order.pay();
    orders.update(order);
    return OrderResponse.from(order);
}

应用里已有 RedissonClient 时直接用它；没有时按 Spring Boot 的 Redis 连接信息（spring.data.redis.*）创建一个单机连接。Sentinel、Cluster 或需要自定义 TLS 时，自己声明 RedissonClient。

聚合锁 ​

方面行为
锁名<前缀>lock:<type>:<id>。id 是对方法参数求值的 SpEL 表达式，类型化标识取原始值
等待注解上的 waitTime，或 ddk.concurrency.lock.wait-time。超时抛 AggregateBusyException
持有不设置持有时间时由 Redisson 的看门狗自动续期，直到方法结束；设置 leaseTime 后到期自动释放，方法超时会打一条警告日志
事务锁包在被标注方法的事务外面：事务开始之前获取，提交或回滚之后释放
重入同一个线程可以再次获取同一把锁

它和乐观锁是什么关系？ 互补，不是替代。乐观锁保证两个写入不会互相覆盖：后提交的那个在 UPDATE ... WHERE version = ? 影响 0 行时失败，见 持久化与聚合还原。但后提交的操作已经把活干完了才被丢弃。聚合锁让冲突的操作排队，第二个操作开始时加载到的就是第一个操作提交后的状态。热点聚合（库存、账户）适合加锁；冲突很少的聚合只用乐观锁就够了。

为什么锁要包住事务？ 顺序反过来的话，锁释放时事务还没提交，下一个操作拿到锁后读到的仍是旧状态，锁就白加了。所以拦截被放在事务通知的外层，有一个测试在 afterCommit 回调里确认锁仍被持有。这也带来一个限制：只有被标注的方法自己开启事务时才成立。调用方已经在事务里时，锁会在外层事务提交之前释放，所以注解要标在最外层的应用服务方法上。

0.7.0 起，这个限制不再靠人记：在事务里面加锁会直接抛出异常，注解和编程式入口都一样。

text
IllegalStateException: Lock [ddk:lock:order:42] is being acquired inside a transaction and would be released before the commit:
acquire the lock first and start the transaction inside it

在事务里重入本线程已经持有的锁不算，那把锁是在事务之外拿到的。有意在事务里面加锁时（例如只想让长事务里的某一步串行），用 ddk.concurrency.lock.inside-transaction=warn 或 ignore 放行。

消息的消费方同时需要锁和幂等消费时，把 Event Starter 的 IdempotentConsumer 放在锁的里面。它在没有事务时会自己开启一个，所以一条语句就得到「锁 → 事务 → 去重登记」的嵌套：

java
aggregateLocks.executeAll("sku", skuIds, () ->
        idempotentConsumer.handle("inventory.order-placed", eventId, () -> reserve(command)));

防重复提交 ​

@Idempotent 在方法执行前用 SET NX 加过期时间登记 <前缀>idempotent:<scope>:<key>，值是一个随机令牌。

  • key 应当由客户端生成，重试时保持不变，例如打开表单时生成的请求 ID。
  • 有效期内用同一个 key 再次调用，抛 DuplicateRequestException。它只拒绝重复请求，不会重放第一次的结果。
  • 方法抛异常时登记被撤销，客户端可以用同一个 key 重试。只撤销本次调用自己的登记：一次调用超过有效期才失败，而期间另一次调用已经登记了同一个 key，它不会删掉别人的登记（比较令牌后再删除）。
  • 撤销登记失败（Redis 不可达）时，调用方拿到的仍是方法自己的异常，Redis 的错误作为 suppressed 附在上面；登记随有效期过期。
  • scope 默认是声明类的全名、方法名和参数类型，例如 com.acme.OrderService#place(PlaceOrderCommand)。不同包里的同名类、重载方法不会共用登记，代理也不改变 scope。
  • 重复请求在加锁之前就被拒绝，不占用锁的等待时间。

它保证的是：有效期内，同一个 key 同一时刻最多放行一次调用，失败的调用可以重试。它不保证副作用恰好执行一次：超过有效期的调用可能和重试重叠，成功的调用也只在有效期内保有登记。支付等必须只发生一次的操作，要靠业务唯一键、数据库状态检查或外部系统自己的幂等键。

0.9 的升级影响

默认 scope 从 OrderService.place 变成了 com.acme.OrderService#place(PlaceOrderCommand)，Redis 的 key 随之改变。发布期间，一个有效期内旧版本登记的请求不会被新版本识别为重复，限流窗口也重新开始计数。需要 key 保持不变时显式设置 scope。

它防的是连点和客户端重试。消费「至少一次」投递的消息要用 事件 starter 的 IdempotentConsumer：它把消息登记和处理逻辑的写入放在同一个数据库事务里，而这里的登记在 Redis 里，和业务事务不是原子的。

限流 ​

java
@RateLimit(limit = 20, period = "1m", key = "#query.tenantId()")
public PageResponse<OrderResponse> search(OrderQuery query) { ... }
  • 每个 key 各有一份额度：一个 period 内最多 limit 次，类型化标识取原始值。不写 key 时整个用例共用一份。
  • 计数放在 Redis 里（Redisson 的 RRateLimiter），所有实例共享。
  • 超过限额时方法不执行，抛 RateLimitedException（RATE_LIMITED），Web starter 返回 429。
  • 它排在 @Idempotent 和 @AggregateLock 之前：被限流的调用不占用请求 key，也不等待锁。
  • 次数和窗口是 Redis key 的一部分：<前缀>rate:<scope>:<limit>/<period>:<key>。Redisson 把限流器的配置存在 Redis 里，改了注解本来会继续沿用旧的速率；把配置放进 key 名后，新版本用的是新 key，旧 key 闲置两个窗口后过期。

它管的是业务限额，例如每个租户的导出次数、每个手机号的短信验证码次数。它在应用内部、请求解析之后才执行，挡不住流量洪峰，不能代替网关层的限流。

配置项 ​

配置项默认值说明
ddk.concurrency.enabledtrue关闭注解与客户端，仅用于本地排查
ddk.concurrency.key-prefixddk:所有 Redis key 的前缀，多个应用共用 Redis 时按应用设置
ddk.concurrency.lock.wait-time3s等待锁的默认时间
ddk.concurrency.lock.lease-time持有锁的默认时间，不设置则由看门狗续期
ddk.concurrency.lock.inside-transactionfail在事务里面加锁时：fail 抛出异常，warn 记警告，ignore 不检查（0.7.0 起）
ddk.concurrency.idempotent.ttl10m请求登记的默认保留时间

实现要点 ​

  1. 只依赖 Redisson 的核心客户端。 Redisson 自己的 Spring Boot starter 会替换 Spring Data Redis 的连接工厂，影响应用里已有的 Redis 用法（包括 DDK 的 Redis 与缓存 starter）。
  2. 没有客户端时直接失败。 容器里没有 RedissonClient 时，调用带注解的方法会抛 IllegalStateException。悄悄地不加锁比报错危险得多。
  3. 拦截顺序靠后置处理器保证。 拦截挂在所有其他后置处理器之后，并插到已有通知之前，所以锁在事务通知的外层。

不适用的场景 / 已知问题 ​

  1. 示例：用户示例不依赖 Redis，暂时没有用到本 starter。
  2. 故障切换：锁的可靠性取决于单个 Redis。主从切换时，还没复制到从节点的锁可能被再次授予，所以聚合上的版本校验不能去掉。原理见 Redis 原子性、事务与锁。
  3. 跨聚合：一次只锁一个聚合。需要同时修改多个聚合时，先确认聚合边界是否合理，见 聚合边界。

文章以 CC BY-NC-SA 4.0 授权 · 代码片段以 MIT 授权

© 2025-present Elijah Du