KEEL · 龙骨 · A CURRICULUM FOR THE AI ERA

04 · 并发控制：分布式锁、悲观锁与乐观锁 — keel 龙骨

这一章回答：多个请求同时改同一份数据时，怎么保证不出错，以及三种锁各自什么时候用。

这一章回答：多个请求同时改同一份数据时，怎么保证不出错，以及三种锁各自什么时候用。

三个层次要分清：进程内的协程/线程互斥（第 02 章的 asyncio.Lock）、跨实例的分布式锁（Redis）、数据库层的行锁与版本控制。混用或用错层级，是并发 bug 最常见的来源。

一、选型总表

手段	保证范围	适用	代价
`asyncio.Lock` / `threading.Lock`	单进程内	进程内共享资源的互斥	分布式下无效
分布式锁（Redis `SET NX EX`）	跨实例	需要"同一时刻只有一个执行者"的任务	依赖 Redis 可用性，需要处理超时与误删
悲观锁（`SELECT ... FOR UPDATE`）	数据库行	冲突频繁、要求强一致的写（资金、库存扣减）	锁等待、可能死锁，吞吐受限
乐观锁（version 字段）	数据库行	读多写少、冲突较少的更新（秒杀、状态流转）	冲突方需要重试

经验判断：冲突频繁用悲观，冲突稀少用乐观；跨服务协调用分布式锁；同一行数据的更新优先交给数据库本身。

二、分布式锁：三个必须做对的点

最小可用实现（Redis SET 扩展参数）：

async def acquire_lock(redis, key, value, expire=5):
    # NX：只在不存在时设置；EX：带过期时间，防止持有者崩溃导致死锁
    return await redis.set(key, value, nx=True, ex=expire)

async def release_lock(redis, key, value):
    # 必须校验 value 再删：否则可能删掉别人加的锁
    script = """
    if redis.call('get', KEYS[1]) == ARGV[1] then
        return redis.call('del', KEYS[1])
    else
        return 0
    end
    """
    return await redis.eval(script, 1, key, value)

三个关键点：

加锁必须带过期时间（EX）：持有者崩溃时锁能自动释放，否则整个业务永久卡死。
释放必须校验唯一值（UUID 等）：否则超时后被别人拿走的锁会被你误删——这是最经典的分布式锁 bug。
释放必须原子执行（Lua 脚本）：GET 再 DEL 两步之间存在窗口。

另外两个常被忽略的问题：

业务执行时间超过锁过期时间：锁自动失效而业务还在跑，出现两个执行者。需要"看门狗"续期，或把过期时间设得足够保守。
Redis 单点/主从切换：故障切换时锁可能丢失。强一致场景应使用 Redlock 或改用数据库锁。大多数业务不需要为这个极端场景付代价，但要知道它存在。

三、悲观锁：让数据库替你排队

async def deduct_with_pessimistic_lock(session, order_id, amount):
    async with session.begin():
        row = await session.execute(
            select(Account).where(Account.id == order_id).with_for_update()
        )
        account = row.scalar_one()
        if account.balance < amount:
            raise InsufficientBalance()
        account.balance -= amount
    # 事务提交时释放行锁

要点：

FOR UPDATE 锁的是命中的行，所以 WHERE 条件必须走索引，否则可能升级为更大范围的锁甚至锁表；
事务要短：任何网络调用都不应出现在持锁事务里；
注意死锁：多表更新要统一顺序；
适用量级参考：写冲突高、QPS 万级以下的场景较合适。

四、乐观锁：让冲突方重试

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    retry=retry_if_exception_type(VersionConflict),
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=5),   # 指数退避，避免重试风暴
    reraise=True,
)
async def reduce_stock(session, sku_id, quantity):
    async with session.begin():
        sku = (await session.execute(select(Sku).where(Sku.id == sku_id))).scalar_one()
        if sku.stock < quantity:
            raise OutOfStock()
        new_version = sku.version + 1
        result = await session.execute(
            update(Sku)
            .where(Sku.id == sku_id, Sku.version == sku.version)   # 核心：版本作为条件
            .values(stock=Sku.stock - quantity, version=new_version)
        )
        if result.rowcount == 0:
            raise VersionConflict()   # 说明这期间被别人改过

要点：

更新语句的 WHERE 必须带上读取时的 version，这是整个机制的关键；
rowcount == 0 是冲突信号，必须处理而不是忽略；
重试要有上限 + 指数退避：立刻重试会把冲突率进一步推高；
适合高并发读多写少（如秒杀场景可达十万级 QPS），无死锁。

常见组合实践：读多写少的热点商品用乐观锁 + 独立库存表（避免与主业务表竞争同一批行锁）；资金类用悲观锁。

动手：可观察结果

产出	判断标准
三种锁的对比实现	各自能跑通同一个"扣库存"场景，并有冲突测试
并发压测数据	100 并发抢同一 SKU：无超卖；记录乐观锁与悲观锁各自的吞吐与失败重试次数
一张选型表	对你项目的每个并发写点标注使用哪种锁及理由

完成标志：在压测下库存不为负、无重复扣减，且能说清为什么这里选了这种锁而不是别的。

故障注入

注入方式	观察
分布式锁释放时不校验 value	制造"超时后被他人获取"的场景，看是否误删别人的锁
释放分两步 `GET` + `DEL`	高并发下是否出现删错锁的情况
持锁期间做网络调用	锁等待时间是否被放大，连接池是否被打满
`FOR UPDATE` 的 WHERE 不走索引	观察锁范围与慢查询，是否出现大面积阻塞
乐观锁不加乐观重试退避（立即重试）	冲突率是否进一步升高，形成活锁
乐观锁更新不带 version 条件	是否出现丢失更新（last write wins）

自测题

分布式锁释放时为什么必须校验 value？分步 GET+DEL 的窗口问题出在哪？
业务执行时间超过锁过期时间会怎样？有哪些应对方式？
SELECT ... FOR UPDATE 在什么条件下会锁表？如何避免？
乐观锁的更新语句里，version 条件放在哪里？漏掉会怎样？
为什么重试要用指数退避而不是立即重试？

进入 keel 阅读