Python asyncio.Lock 怎样让共享计数更新保持原子性

多个协程同时更新一个共享计数时，问题通常不在“加法”本身，而在读、改、写不是一个不可分割的动作。只要两个任务在读取旧值后交错执行，后写入的结果就可能覆盖前一个任务的更新。asyncio.Lock 的作用是让同一事件循环中的协程按互斥方式进入这段临界区。

官方地址：https://docs.python.org/3/library/asyncio-sync.html

这篇只讨论异步任务内的共享计数保护：如何确定锁的范围、怎样可靠释放，以及什么时候不该继续扩大锁的职责。它不把 asyncio.Lock 当作操作系统线程锁。

先把共享更新看成一个受约束的资源

可以把计数器看作一个只有一个写入口的共享资源。每个协程都需要先读取当前值，再加一，最后写回；如果这三个动作之间允许另一个任务插入，最终值就未必等于任务数量。这个场景的约束很明确：临界区要短，锁只服务于同一个事件循环中的协程，真正的 I/O 或 CPU 密集工作不要放进去。

import asyncio

async def unsafe_increment(state):
    # 让出事件循环，模拟读改写之间存在可交错的异步边界。
    old_value = state["count"]
    await asyncio.sleep(0)
    state["count"] = old_value + 1

async def main():
    state = {"count": 0}
    # 多个任务可能读到同一个旧值，最后一次写入会覆盖前面的更新。
    await asyncio.gather(*(unsafe_increment(state) for _ in range(100)))
    print(state["count"])

asyncio.run(main())

这里的 sleep(0) 不是业务修复，而是把任务切换点显式放在读和写之间，帮助理解为什么“单线程事件循环”也不自动意味着共享状态更新安全。实际代码中的网络请求、队列操作或其他 await 都可能形成类似的切换边界。

用 async with 把读改写放进同一个临界区

官方文档推荐用 async with lock 管理锁。对于共享计数，锁必须覆盖完整的读改写，而不是只包住最后一行赋值；否则两个任务仍然可能在读取阶段拿到同一个旧值。

import asyncio

async def safe_increment(state, lock):
    # 读、改、写属于同一个临界区，其他协程会在入口等待。
    async with lock:
        state["count"] += 1

async def main():
    state = {"count": 0}
    lock = asyncio.Lock()

    # 所有任务共享同一把锁，不能在每个任务内部重新创建锁。
    await asyncio.gather(*(safe_increment(state, lock) for _ in range(100)))
    print(state["count"])

asyncio.run(main())

判断写法是否正确，可以只看两个问题：所有访问共享计数的任务是否拿到同一个锁，以及锁是否覆盖了完整的状态转换。若每个任务都创建自己的 asyncio.Lock()，那些锁之间没有共同的互斥关系，效果等同于没有加锁。

锁的范围要短，等待动作要放在外面

锁保护的是共享状态，不是整个业务流程。假设任务要先请求远端接口，再把响应写入内存计数，通常应把网络等待放在锁外，只在真正修改共享对象的几行代码上加锁。这样既减少排队，也避免一个慢请求让所有更新者一起等待。

import asyncio

async def fetch_and_count(state, lock, fetch):
    # 外部等待不占用共享状态的锁，避免把慢 I/O 变成全局排队。
    result = await fetch()

    # 只在合并结果这一小段时间内互斥访问共享状态。
    async with lock:
        state["count"] += result

async def fake_fetch():
    # 这里只模拟异步来源，真实项目可以替换成 HTTP 或数据库调用。
    await asyncio.sleep(0.01)
    return 1

这是一种取舍：锁越小，吞吐通常越好，但必须确认锁外的计算不会读取一个稍后失效的共享快照。如果业务要求“读取、等待、校验、提交”整体不可被插入，就需要重新设计状态机，而不是盲目把所有步骤塞进锁里。

async with 和 try/finally 怎样选择

async with 适合绝大多数临界区，因为离开代码块时会自动释放锁。需要把获取、释放封装进更复杂的控制流程时，也可以显式调用 acquire()，但必须用 try/finally 保证异常或任务取消时不会遗留锁。

async def manual_increment(state, lock):
    # 手动获取锁时，finally 是释放锁的最低保障。
    await lock.acquire()
    try:
        state["count"] += 1
    finally:
        # 即使临界区抛出异常或任务被取消，也要释放锁。
        lock.release()

两种方式表达的是同一个资源边界：先获得互斥权，再修改共享状态，最后归还互斥权。不要在一个函数里无条件重复释放同一把锁；未持有锁时调用 release() 会抛出 RuntimeError。

几个方案怎么选：Lock、Semaphore 还是线程锁

选型时先看资源约束，而不是先换一个看起来更复杂的同步原语。

官方文档还说明，asyncio 同步原语本身不接受超时参数；如果确实需要等待上限，应在调用层用 asyncio.wait_for() 等方式包住等待，并单独决定超时后的业务回滚策略。

落地前记住这四个边界

• 同一把锁才有互斥关系。锁应由共享资源的所有协程共同持有，不能每次调用临时创建。
• 临界区覆盖完整状态转换。对计数器来说，读、改、写不能拆开保护。
• 锁内不要做慢 I/O。先完成外部等待，再锁住最短的合并动作。
• 不要跨线程套用。asyncio.Lock 解决的是事件循环内的协程协作，不是线程间同步。

总结：先定义资源，再定义锁的边界

让共享计数更新保持原子性，核心不是给代码贴上“并发安全”标签，而是明确一段短小、完整、可释放的临界区：所有协程共享同一把 asyncio.Lock，用 async with 包住读改写，遇到手动控制时用 try/finally 释放。把网络等待和线程同步排除在这把锁的职责之外，代码才会既正确又容易扩展。

相关问题

单线程事件循环为什么还需要 Lock？因为协程会在 await 处交错执行，读改写之间的状态可能被另一个任务改变。

可以只给共享计数的赋值语句加锁吗？不建议。需要把读取旧值、计算新值和写回动作放在同一个临界区。

asyncio.Lock 能限制同时运行的任务数量吗？它只允许一个协程进入临界区；需要多个并发名额时应考虑 asyncio.Semaphore。