多线程并发操作数据库时的数据一致性问题咨询

阿华AIGC实验室

2026-5-26

解决多线程并发更新数据库的一致性问题（丢失更新场景）

Hey, this is a classic Lost Update scenario—super common when multiple threads read the same database row, perform calculations based on the initial value, and write back later. In your case, T3 will overwrite T2's 60 result with whatever it computes from the original 50, completely ignoring the updates from T1 and T2. Let’s break down the most practical solutions for this problem:

1. 乐观锁（Optimistic Locking）

这是应用最广泛的方案之一，非常适合并发冲突不频繁的场景。

核心原理：给数据表加一个版本号字段（比如version）或者时间戳字段（比如update_time）。读取数据时同时获取版本号，更新时带上这个版本号作为条件——只有数据库中的版本号和你读取的完全一致，更新才会生效。如果更新失败，说明已有其他线程修改过数据，需要重新读取最新值并重试。
示例SQL：
读取操作：
```
SELECT value, version FROM your_table WHERE id = ?;
```
更新操作（以T1给值加5为例）：
```
UPDATE your_table 
SET value = value + 5, version = version + 1 
WHERE id = ? AND version = ?;
```
执行更新后要检查受影响行数，如果是0，就触发重试逻辑，重新读取最新数据再操作。
优势：无需加锁，性能开销小；适合读多写少、冲突率低的系统。
劣势：如果并发冲突频繁，会导致大量重试，影响业务效率。

2. 悲观锁（Pessimistic Locking）

适合并发冲突频繁的场景，读取数据时直接加排他锁，阻止其他线程修改，直到当前线程完成更新释放锁。

核心原理：读取数据时使用数据库的排他锁机制，其他线程尝试读取或修改该数据时会被阻塞，直到当前线程提交事务释放锁。
示例SQL（数据库语法略有差异）：
MySQL中使用FOR UPDATE：
```
SELECT value FROM your_table WHERE id = ? FOR UPDATE;
```
执行这条语句后，目标行被加排他锁，其他线程执行相同的锁读取或更新操作时会进入阻塞状态，直到当前事务提交。
注意事项：必须在事务中使用，否则锁会立即释放；要控制事务执行时长，避免长时间占用锁导致其他线程等待过久。
优势：冲突处理逻辑简单，无需重试；适合写多、冲突率高的系统。
劣势：加锁会导致线程阻塞，性能开销大；存在死锁风险（多个线程互相等待对方释放锁）。

3. 原子更新语句（Atomic Update Statements）

把「读取-计算-更新」的逻辑直接放到数据库的单条SQL语句中，利用数据库的原子性保证操作不会被打断，从根源避免读取旧值的问题。

核心原理：让数据库端直接完成基于最新值的计算和更新，客户端不需要读取初始值，自然不会出现基于旧值覆盖更新的情况。
示例SQL：
不管每个线程要加多少值，直接执行：
```
UPDATE your_table SET value = value + ? WHERE id = ?;
```
T1传入5，T2传入10，T3传入自己要加的数值——数据库会自动基于当前最新的value进行累加，完全不会出现覆盖问题。
优势：实现最简单，不需要额外字段（比如版本号），性能优异（数据库原生支持的原子操作）。
劣势：仅适用于简单的增减操作，如果更新逻辑复杂（比如需要关联其他表数据、做复杂计算），这种方式就不适用了。

4. 应用层分布式锁（Application-Level Distributed Locks）

如果你的应用是分布式部署（多实例），或者需要跨多个数据库保证一致性，可以用分布式锁来确保同一时间只有一个线程能操作目标数据。

核心原理：借助Redis、ZooKeeper等组件实现分布式锁，线程在操作数据前必须先获取锁，获取成功后才能执行读写操作，完成后释放锁；获取失败则等待或重试。

示例伪代码：

lock_key = f"lock:your_table:{row_id}"
# 设置3秒锁超时，避免线程崩溃后锁永久占用
if redis.acquire_lock(lock_key, expire=3000):
    try:
        # 读取最新值
        value = db.fetch_one("SELECT value FROM your_table WHERE id = ?", row_id)
        # 计算新值
        new_value = value + delta
        # 更新数据
        db.execute("UPDATE your_table SET value = ? WHERE id = ?", new_value, row_id)
    finally:
        # 确保即使出错也能释放锁
        redis.release_lock(lock_key)
else:
    # 处理锁获取失败：重试或返回错误给用户
    retry_operation_or_return_error()