分布式系统中数据库写入竞态条件的解决方案探究

问题场景 #

某费用报告服务支持多实例部署，员工创建报告时，服务会先读取该员工已有报告数量，若达到上限（示例为5份）则拒绝请求。但当员工快速提交多份请求时，多实例并行处理会触发竞态条件：比如两个实例同时读取到员工已有4份报告，随后各自写入新报告，最终导致报告数突破上限到6份，违反限制规则。

可行解决方案 #

1. 数据库行级悲观锁 #

在读取员工报告计数时，对该员工的关联记录加排他锁，确保同一时刻只有一个服务实例能完成“读取-判断-写入”的完整流程。

• 具体实现：用SELECT ... FOR UPDATE语句锁定目标行，其他实例必须等待锁释放后才能操作。比如单独维护一张员工报告统计表（employee_report_stats），操作逻辑如下：

  BEGIN TRANSACTION;
  -- 锁定该员工的统计行，防止其他实例同时读取
  SELECT report_count FROM employee_report_stats WHERE employee_id = ? FOR UPDATE;
  IF report_count < 5 THEN
      INSERT INTO expense_reports (...) VALUES (...);
      UPDATE employee_report_stats SET report_count = report_count + 1 WHERE employee_id = ?;
  END IF;
  COMMIT;
  

• 注意事项：事务隔离级别设为REPEATABLE READ避免幻读，尽量缩短锁持有时间，减少对并发性能的影响。

2. 乐观锁机制 #

不主动加锁，而是通过版本号校验来检测数据是否被其他实例修改，冲突时重试或拒绝请求。

• 具体实现：在employee_report_stats表中加一个version字段，读取时同时获取计数和版本号，更新时校验版本号是否匹配：

  -- 读取当前计数和版本
  SELECT report_count, version FROM employee_report_stats WHERE employee_id = ?;
  IF report_count < 5 THEN
      -- 只有版本号匹配才允许更新计数
      UPDATE employee_report_stats 
      SET report_count = report_count + 1, version = version + 1 
      WHERE employee_id = ? AND version = ?;
      -- 检查更新影响行数，0代表已被其他实例修改
      IF affected_rows == 0 THEN
          -- 处理冲突：可以重试几次，或者直接返回“请求繁忙”
      ELSE
          INSERT INTO expense_reports (...) VALUES (...);
      END IF;
  END IF;
  

• 优势：无锁等待，适合并发高、冲突少的场景；冲突时通过重试保证最终一致性。

3. 原子化数据库操作 #

把“计数校验+写入报告”合并成一个原子操作，利用数据库的原子性避免竞态。

• 方式一：用INSERT ... SELECT带条件判断，直接在插入时校验上限：

  INSERT INTO expense_reports (employee_id, ...)
  SELECT ?, ...
  FROM dual
  WHERE (SELECT COUNT(*) FROM expense_reports WHERE employee_id = ?) < 5;
  

  插入后检查影响行数，若为0说明已达上限，直接拒绝请求。
• 方式二：用存储过程封装整个逻辑，确保所有操作在一个事务里原子执行。

4. 分布式锁 #

借助Redis、ZooKeeper等工具实现分布式锁，给每个员工的创建请求加锁，同一时间只有一个实例能处理该员工的请求。

• 实现要点：
  • 锁的key用employee_id，保证粒度精准，不影响其他员工的并发
  • 设置锁超时时间，防止实例崩溃导致锁永久占用
  • 锁获取失败时，直接返回“请求过于频繁”或重试
• 示例（Redis锁伪代码）：

  lock_key = f"expense_report_lock:{employee_id}"
  # 尝试获取锁，超时3秒
  if redis_client.set(lock_key, "locked", nx=True, ex=3):
      try:
          report_count = db.query("SELECT COUNT(*) FROM expense_reports WHERE employee_id = ?", employee_id)
          if report_count < 5:
              db.execute("INSERT INTO expense_reports (...) VALUES (...)")
      finally:
          # 无论成功失败都释放锁
          redis_client.delete(lock_key)
  else:
      return {"error": "请求太频繁啦，稍后再试吧"}
  

5. 请求串行化处理 #

用消息队列把同一员工的请求串行化，避免并行处理。

• 实现方式：把所有创建请求发到消息队列，按employee_id做分区，同一个员工的请求只会被同一个消费者线程处理，天然保证串行执行，从根源上避免竞态。
• 额外好处：还能实现流量削峰，避免突发请求压垮数据库。

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