Spring WebFlux中如何处理read-modify-write竞态条件？

方案1：数据库原子更新（优先推荐） #

这是性能最高、无竞态风险的实现方式，直接依托数据库单步更新的原子性规避read-modify-write问题，无需额外加锁或重试逻辑：

1. 首先在CategoryRepository中新增原子计数更新方法，以R2DBC为例：

@Modifying
@Query("UPDATE category SET number_of_items = number_of_items + 1 WHERE id = :categoryId")
Mono<Integer> incrementItemCount(Long categoryId);

2. 调整原有业务代码，省去查询分类实体的步骤：

itemMono.flatMap(itemRepository::save)
        .flatMap(item -> categoryRepository.incrementItemCount(item.getCategoryId()))
        // 建议补充异常处理逻辑避免静默失败
        .doOnError(e -> log.error("更新分类商品计数失败", e))
        .subscribe();

方案2：乐观锁+重试（适用于必须读取分类实体做其他业务处理的场景） #

如果你的业务逻辑要求必须先拿到完整的分类实体做其他字段修改，可通过乐观锁控制并发冲突：

1. 给Category实体添加版本号字段，开启Spring Data乐观锁支持：

@Data
@Table("category")
public class Category {
    // 其他原有字段
    @Version
    private Long version;
}

2. 调整业务代码，捕获乐观锁冲突异常后重试：

itemMono.flatMap(itemRepository::save)
        .flatMap(item -> categoryRepository.findById(item.getCategoryId())
                .map(category -> category.toBuilder()
                        .numberOfItems(category.getNumberOfItems() + 1)
                        // 其他自定义业务修改
                        .build())
                .flatMap(categoryRepository::save)
                // 仅对乐观锁异常做最多3次重试，可根据业务场景调整重试次数
                .retryWhen(Retry.max(3)
                        .filter(e -> e instanceof OptimisticLockingFailureException))
        )
        .doOnError(e -> log.error("更新分类信息失败", e))
        .subscribe();

注意：该方案仅适合并发冲突率较低的场景，冲突过高时多次重试会额外消耗性能。

方案3：分布式锁（适用于并发冲突极高的场景） #

如果业务同时要求读取分类实体、且并发冲突率很高不适合用乐观锁重试，可引入Redis等分布式锁串行化执行分类更新逻辑，以Redisson反应式API为例：

itemMono.flatMap(itemRepository::save)
        .flatMap(item -> {
            RLockReactive lock = redissonReactiveClient.getLock("category:lock:" + item.getCategoryId());
            return lock.lock(10, TimeUnit.SECONDS) // 锁10秒自动过期避免死锁
                    .then(Mono.defer(() -> categoryRepository.findById(item.getCategoryId())
                            .map(category -> category.toBuilder().numberOfItems(category.getNumberOfItems() + 1).build())
                            .flatMap(categoryRepository::save)
                    ))
                    .doFinally(signalType -> lock.unlock().subscribe());
        })
        .doOnError(e -> log.error("更新分类信息失败", e))
        .subscribe();

注意事项 #

• 不要直接调用subscribe()后忽略结果，务必补充异常捕获、日志记录逻辑，避免业务失败无感知。
• 如果要求商品保存和计数更新的强一致性，需要引入响应式事务管理，可通过@Transactional注解或TransactionalOperator实现事务包裹。
• 如业务允许短时间内的计数不一致，可额外增加定时任务，定期通过COUNT聚合查询修正分类商品数量，保障最终一致性。

内容的提问来源于stack exchange，提问作者Mohamad Sharifi