最近在帮团队做技术评审的时候一个看似基础的问题居然引发了不少讨论同样是加锁threading.Lock和asyncio.Lock到底能不能混着用很多人能说出一个是同步、一个是异步但深入问下去比如为什么async with不能用在threading.Lock上、为什么协程里不能用普通的threading.Lock来保护共享状态很多人就说不清楚了。这篇文章就把这两种锁从底层原理到实战用法掰开揉碎讲一遍看完你不仅能分清区别还能避掉不少线上才会踩的坑。1. 两种并发模型的底层差异决定了锁的本质不同先说结论threading.Lock和asyncio.Lock的服务对象完全不同前者服务的是操作系统线程后者服务的是事件循环里的协程任务。这不是换个名字那么简单两者的调度模型、阻塞语义、性能特征都是两个世界。1.1 线程锁的核心让操作系统来帮忙停车threading.Lock底层调用的是操作系统提供的互斥量mutex。当一个线程执行lock.acquire()的时候如果锁已经被其他线程占用这个线程会真正地被操作系统挂起也就是从运行状态切换到阻塞状态CPU 会去执行其他线程。这个过程叫做线程上下文切换成本相当高——需要保存当前线程的寄存器状态、程序计数器、栈指针然后加载下一个线程的上下文整套流程是纳秒级到微秒级的花销。你从业务代码的角度来看threading.Lock的acquire()是一个硬阻塞调用一旦阻塞这个线程在当前时间片内什么事都干不了只能等操作系统重新调度它。这种设计在线程密集型的场景中没问题但在大量连接需要同时处理的网络编程场景中一个线程对应一个连接的做法会迅速耗尽系统资源——这也是为什么异步编程在 I/O 密集型场景中崛起的核心原因之一。1.2 协程锁的核心让事件循环来协调让路asyncio.Lock的底层不是操作系统锁它本质上是一个基于 Future 的调度器协作机制。当一个协程执行await lock.acquire()的时候如果锁不可用这个协程会挂起但挂起的方式完全不同——它把控制权主动交还给了事件循环事件循环立刻可以去跑其他就绪的协程任务直到持有锁的协程释放锁事件循环才会把排在等待队列里的协程唤醒。这个挂起-让路-唤醒的过程就是协程的核心优势没有线程上下文切换没有内核态切换纯粹在用户态由事件循环调度。我测试过一个极端场景asyncio.Lock在十万次竞争下的开销远低于threading.Lock在同样次数下的开销前提是这些竞争发生在同一个事件循环内。我之前做过一个简单的基准测试创建 10000 个协程去抢一把锁asyncio.Lock整体耗时在毫秒级而用threading.Lock配合线程池模拟同样规模耗时直接到了秒级栈内存占用更是天差地别。1.3 一个类比帮你彻底记住把两种锁类比成不同场景下的排队机制threading.Lock像银行柜台——每个排队的人都是一个实体的线程排在后面的人什么都做不了只能在原地等柜台CPU只能服务当前正在办理的那个人。asyncio.Lock像医院叫号系统——协程就像拿着号的病人没叫到号的时候你可以去喝水、看手机、做检查执行其他协程等叫到号了再过去办理。这个类比虽然不完全精确但对理解为什么异步里不能有阻塞调用非常有帮助如果你在协程里调用threading.Lock的acquire()相当于叫号系统里一个病人站在窗口前干等整个叫号系统被你堵死了。2. 关键API差异context manager、超时与取消语义两种锁虽然都提供了acquire()和release()但在具体 API 语义上有不少细微差别这些差别恰恰是写代码时最容易出问题的地方。2.1 context manager 的真相不只是语法糖两者都支持with或async with语法但内部机制完全不同。threading.Lock实现的是__enter__和__exit__这是 Python 对象协议的一部分with语句会直接调用这两个方法。asyncio.Lock实现的是__aenter__和__aexit__这是异步上下文管理器协议必须配合async with使用。我第一次写异步代码时犯过这样的错误拿着threading.Lock直接在协程函数里写async with lock:结果直接抛出AttributeError因为threading.Lock根本没有__aenter__方法。反过来把asyncio.Lock放在同步函数里写with lock:也会直接报错。这里有一个很容易被忽略的细节asyncio.Lock的acquire()在拿到锁之前不保证公平性也就是说先等待的协程不一定会先获取锁。Python 官方文档里明确提过asyncio.Lock的获取顺序取决于事件循环的调度实际上在 CPython 的早期实现里甚至出现过饥饿问题不过后续版本通过 FIFO 队列做了改进。而threading.Lock在 CPython 中默认也是“非公平”的操作系统调度决定的顺序但因为它本身是阻塞式的这种非公平性对业务的感知不那么明显。2.2 超时参数一个支持一个不支持threading.Lock.acquire(timeout...)是支持的超过指定时间如果还没拿到锁返回False你可以在代码里决定是重试还是放弃import threading lock threading.Lock() if lock.acquire(timeout5.0): try: # 拿到了锁做事情 pass finally: lock.release() else: # 5秒没拿到锁 print(获取锁超时)但asyncio.Lock.acquire()不支持 timeout 参数调用await lock.acquire(timeout5)会直接抛出TypeError。想要超时控制的话需要借助asyncio.wait_for来包装import asyncio lock asyncio.Lock() async def try_acquire(): try: await asyncio.wait_for(lock.acquire(), timeout5.0) except asyncio.TimeoutError: print(获取锁超时处理冲突逻辑) return try: # 拿到了锁做事情 pass finally: lock.release()注意wait_for的超时取消只能取消acquire()的等待过程一旦协程已经拿到了锁wait_for的超时并不会自动释放锁。这点很重要否则会出现锁泄漏——锁永远没人释放其他协程全部卡死。2.3 取消语义协程锁的隐藏陷阱asyncio.Lock还有一个threading.Lock完全不具备的特性它需要考虑任务取消Task Cancellation的影响。当一个协程正在await lock.acquire()时如果外部对这个任务发起了cancel()协程会抛出CancelledError此时锁的等待状态会被正确清理不会污染锁的内部状态。但如果你已经拿到了锁然后业务代码里遇到了异常或取消锁不会自动释放你必须用try/finally来保证释放。这里有个和线程锁类似的经典错误async def bad_example(): async with lock: await some_io_operation() # 如果这里抛异常锁依然会释放 # 但如果锁是在 enter 之前拿到的就必须小心了用async with能保证异常时释放锁这个和with类似。真正容易出问题的是这种写法lock asyncio.Lock() async def risky(): await lock.acquire() # 没有用 async with手动管理 await do_something() lock.release()如果do_something()抛了异常release()永远不会执行锁就泄漏了。所以说实话在绝大多数业务代码里我更推荐直接使用async with lock:这种写法让 Python 帮你兜底。3. 核心区别锁的作用域决定了你该怎么选很多人会忽略一个关键点threading.Lock的锁状态是线程间全局可见的而asyncio.Lock的锁状态只在同一个事件循环内可见。这个区别在最极端的场景下会直接决定程序能不能跑对。3.1 跨线程 vs 跨协程能混用吗直接回答不推荐混用而且大部分情况下混用会出大问题。设想一个场景事件循环中的协程 A 持有了一个threading.Lock同时线程池中的线程 B 也去抢同一把锁。如果 A 和 B 操作的是同一个共享资源这听起来好像“也合理”但问题在于A 拿锁之后如果遇到了真实的 I/O 等待比如await asyncio.sleep()它会主动让出控制权而事件循环在等待期间可能调度其他协程任务如果那些协程任务又试图获取同一把threading.Lock就会发生阻塞事件循环的行为把整个异步循环卡死。反过来如果你在一个协程里await asyncio.Lock但同一个锁还需要被线程池里的同步任务使用这是不能直接用的——asyncio.Lock只能被事件循环内的协程使用你在线程池里调用asyncio.Lock.acquire()不是一个 awaitable 对象在线程里使用的问题而是语义上就不成立。3.2 真实场景选型表场景推荐锁原因纯多线程程序IO 共享变量threading.Lock操作系统线程互斥阻塞等待单线程 异步 I/O 密集任务asyncio.Lock非阻塞协作调度不卡事件循环多个事件循环共享资源推荐使用threading.Lock包裹资源访问或者用外部进程锁asyncio.Lock只对同一循环内协程可见混合场景异步代码调用同步阻塞库用asyncio.to_thread配合threading.Lock保持线程安全避免阻塞事件循环需要跨进程互斥multiprocessing.Lock或第三方库锁进程级互斥线程锁和协程锁都无效这张表是我们在实际项目中总结出来的选型原则核心就一句话锁的作用域必须大于等于共享资源的访问作用域。如果共享资源会被多个线程访问就用线程锁如果只在同一个事件循环里被协程访问就用协程锁。3.3 混合场景的正确姿势最让人头疼的是“异步框架 同步阻塞库”的组合。比如你用httpx的异步客户端没问题但有一天接入了一个只有同步 SDK 的内部服务你不能直接在协程里调用await asyncio.Lock同时又在同步代码里用同一个asyncio.Lock。正确做法是把同步调用丢到线程池用threading.Lock来保护共享状态。我之前做某个内部数据同步项目的时候就是用了这样的模式import asyncio import threading import time import random # 共享状态 shared_cache {} thread_lock threading.Lock() def sync_io_task(key): # 模拟同步阻塞I/O time.sleep(random.uniform(0.1, 0.5)) with thread_lock: value shared_cache.get(key, 0) shared_cache[key] value 1 return shared_cache[key] async def async_worker(key): # 在线程池中执行同步任务 result await asyncio.to_thread(sync_io_task, key) return result async def main(): tasks [async_worker(fkey-{i % 3}) for i in range(30)] results await asyncio.gather(*tasks) print(final shared_cache:, shared_cache) asyncio.run(main())这个模式的实际意义在于把同步阻塞 I/O 放到线程池去执行threading.Lock只保护临界区的共享状态访问不阻塞事件循环。而asyncio.Lock在这种情况下根本用不了因为锁的作用域覆盖不到线程池里的同步代码。4. 实操演示与代码重构从线程版到协程版直接上一段完整的对比代码自己照着跑一遍比看十遍文档都有用。4.1 线程版模拟并发任务交替更新计数器import threading import time counter 0 lock threading.Lock() def worker(): global counter for _ in range(1000): with lock: # 模拟一段短小的临界区操作 temp counter time.sleep(0.0001) counter temp 1 threads [] for _ in range(10): t threading.Thread(targetworker) t.start() threads.append(t) for t in threads: t.join() print(threading counter:, counter)如果你把time.sleep(0.0001)挪出with lock块最终计数结果会小于10000因为多个线程同时读取旧值然后写回丢了更新。这就是典型的竞态条件race condition。4.2 协程版同样的逻辑用 asyncioimport asyncio counter 0 lock asyncio.Lock() async def worker(): global counter for _ in range(1000): async with lock: temp counter # 模拟异步I/O等待但注意锁已经握住了 await asyncio.sleep(0.0001) counter temp 1 async def main(): await asyncio.gather(*(worker() for _ in range(10))) print(asyncio counter:, counter) asyncio.run(main())这两份代码行为几乎完全等价都保证了counter的最终结果为10000。区别在于线程版在临界区里sleep的时候占用的是一条真实的线程资源即使它什么都不做线程被挂起操作系统得调度别的线程。协程版在await asyncio.sleep()的时候事件循环可以继续执行其他协程任务没有“线程被占据”的问题。不过要提醒一句协程版的await asyncio.sleep()放在锁里面等于在临界区里暂停。虽然事件循环不会卡死但锁的持有时间拉长了其他等待锁的协程会排在后面所以临界区的代码应该尽量短小不管是同步还是异步都一样。4.3 从线程版重构到协程版的完整案例现实中的重构往往比上面复杂得多。我在一个模拟场景里处理过一个实际需求原本是 100 个线程并发去调一个第三方 HTTP 接口每个线程访问同一个内存缓存来记录请求计数。后来发现线程开销太大而且第三方接口的响应等待占了大量时间于是想着改成异步方案。初期版本用线程锁import requests import threading cache {} lock threading.Lock() def fetch(url): resp requests.get(url) with lock: cache[url] cache.get(url, 0) len(resp.content)重构为异步版本时很多人会发现requests是同步阻塞库不能直接在协程里用。这时候有两条路用asyncio.to_thread(requests.get, url)把同步请求放到线程池锁仍然用threading.Lock保护缓存。把requests换成httpx.AsyncClient或aiohttp这样全部走异步锁换成asyncio.Lock。推荐方案 2只有彻底异步化才能利用事件循环的优势。但换锁的时候千万别忘了把所有并发入口都改成 async with否则一部分协程用asyncio.Lock另一部分还是threading.Lock共享资源保护就是空的。5. 踩坑实录与排查指南这些坑我基本都踩过一遍这块的实战价值最高。直接把我在实际项目里遇到的几类问题整理出来你对照排查就行。5.1 死锁异步环境下的 串行阻塞先看一个典型的异步死锁场景import asyncio lock asyncio.Lock() async def task_a(): async with lock: await task_b() # 直接调用另一个协程但这个协程也要抢同一种锁 async def task_b(): async with lock: print(never get here) asyncio.run(task_a())执行这个代码程序永远不会打印never get here直接卡死在锁等待上。注意这不是真的死锁而是单事件循环下的锁重入问题——task_a持有锁时调用task_b而task_b尝试获取同一把锁但asyncio.Lock是非可重入的non-reentrant于是永远等待。线程锁也有类似场景一个线程里with lock:再进入同一把lock也会死锁。但异步版本的坑更隐蔽因为await task_b()看起来像函数调用很容易被忽略。排查出这个问题的时候我们在测试环境里直接看到事件循环挂起所有协程都不再输出任何日志一时很难定位。后来靠asyncio.wait_for给所有锁获取加上超时问题才暴露出来——超时异常里直接能看到是哪条协程、哪个文件哪一行等待了哪把锁。5.2 锁泄漏等待方迟迟拿不到锁任务积压asyncio.Lock的锁泄漏通常发生在「手动 acquire 忘记 release」或者「异常路径没有释放」的场景。举个例子lock asyncio.Lock() async def bad_process(): await lock.acquire() try: await call_external_api() # 这里抛异常 finally: pass # 忘记释放锁这个call_external_api()抛异常后finally里没释放锁锁就永远held。后续所有协程在acquire()这里排队时间越久积累的任务越多最终导致内存压力增大甚至进程重启。排查这种问题有个土办法找到锁的关键 waiters 数量做监控。遇到循环调度的 defer 就说明锁泄漏了# 在需要调试的地方插入日志 if lock._waiters: print(flock still has {len(lock._waiters)} waiters)不过我不推荐直接访问私有属性生产环境里去搞可以在锁的 release 处加日志或者用一个简易的计数包装器class LockWithMetrics(asyncio.Lock): def __init__(self): super().__init__() self.acquired_count 0 self.release_count 0 async def acquire(self): await super().acquire() self.acquired_count 1 def release(self): super().release() self.release_count 1监控acquired_count和release_count是否始终平衡如果发散的 lock 增长就说明有泄漏。5.3 性能陷阱临界区里做 I/O 导致吞吐下降可能有人觉得协程锁本身够快于是把大范围的 I/O 操作都放进临界区结果发现整体吞吐急剧下降。原因很简单虽然asyncio.Lock不会阻塞线程但同一把锁在同一时间只允许一个协程进入临界区如果临界区里有一个 300ms 的 HTTP 请求那么每秒最多只能处理约 3 个任务其他请求全部在排队。在这种场景下锁的粒度比锁本身的性能更重要。我的建议是尽量缩短临界区代码只保护真正需要互斥的共享变量读写。把 I/O 操作移出锁外通过“预先取数据、锁内校验/更新”的方式来减少锁持有时间。如果读多写少考虑用asyncio.Semaphore控制并发而不一定用锁互斥。5.4 混用测试ThreadPoolExecutor asyncio.Lock还有一种很隐蔽的坑就是在线程池任务里用了asyncio.Lock。我见过有同事这么写import asyncio from concurrent.futures import ThreadPoolExecutor lock asyncio.Lock() def sync_worker(): # 在线程里尝试用异步锁完全错误 lock.acquire() ...这种写法的结果是acquire()返回了一个 coroutine 对象完全没有执行锁获取当然更没有保护作用。更离谱的是如果你在协程里调用threading.Lock.acquire()虽然能拿到锁但会阻塞事件循环。调试这种 bug 的时候根本找不到具体的抛错信息只有通过性能下降或任务卡死才能看出来。6. 最终经验分享我在实际项目里已经养成了一个习惯在一个项目开始之前先明确“共享资源会被谁访问”——是多个操作系统线程、多个协程还是跨进程。把这个问题的答案写进设计文档再决定使用哪种锁。频繁踩坑之后发现大多数“锁用错了”的问题不是因为不了解锁的 API而是因为对并发边界不够清晰。用asyncio.Lock最核心的心得就是让临界区极短让它内部的 I/O 尽量少用threading.Lock的核心心得则是别在协程里用它也不要在事件循环线程里让它产生真实的阻塞等待。至于哪些代码应该用哪种锁其实只要记住锁的作用域必须覆盖所有可能的访问路径大部分问题都能提前绕开。再补一个小工具层面的建议无论是哪种锁调试期最好都给所有acquire()加上超时机制哪怕是异步锁配合asyncio.wait_for也行。线上没有超时的锁等待是排查死锁时最痛苦的体验加上超时后问题定位会轻松一个数量级。
最近在帮团队做技术评审的时候一个看似基础的问题居然引发了不少讨论同样是加锁threading.Lock和asyncio.Lock到底能不能混着用很多人能说出一个是同步、一个是异步但深入问下去比如为什么async with不能用在threading.Lock上、为什么协程里不能用普通的threading.Lock来保护共享状态很多人就说不清楚了。这篇文章就把这两种锁从底层原理到实战用法掰开揉碎讲一遍看完你不仅能分清区别还能避掉不少线上才会踩的坑。1. 两种并发模型的底层差异决定了锁的本质不同先说结论threading.Lock和asyncio.Lock的服务对象完全不同前者服务的是操作系统线程后者服务的是事件循环里的协程任务。这不是换个名字那么简单两者的调度模型、阻塞语义、性能特征都是两个世界。1.1 线程锁的核心让操作系统来帮忙停车threading.Lock底层调用的是操作系统提供的互斥量mutex。当一个线程执行lock.acquire()的时候如果锁已经被其他线程占用这个线程会真正地被操作系统挂起也就是从运行状态切换到阻塞状态CPU 会去执行其他线程。这个过程叫做线程上下文切换成本相当高——需要保存当前线程的寄存器状态、程序计数器、栈指针然后加载下一个线程的上下文整套流程是纳秒级到微秒级的花销。你从业务代码的角度来看threading.Lock的acquire()是一个硬阻塞调用一旦阻塞这个线程在当前时间片内什么事都干不了只能等操作系统重新调度它。这种设计在线程密集型的场景中没问题但在大量连接需要同时处理的网络编程场景中一个线程对应一个连接的做法会迅速耗尽系统资源——这也是为什么异步编程在 I/O 密集型场景中崛起的核心原因之一。1.2 协程锁的核心让事件循环来协调让路asyncio.Lock的底层不是操作系统锁它本质上是一个基于 Future 的调度器协作机制。当一个协程执行await lock.acquire()的时候如果锁不可用这个协程会挂起但挂起的方式完全不同——它把控制权主动交还给了事件循环事件循环立刻可以去跑其他就绪的协程任务直到持有锁的协程释放锁事件循环才会把排在等待队列里的协程唤醒。这个挂起-让路-唤醒的过程就是协程的核心优势没有线程上下文切换没有内核态切换纯粹在用户态由事件循环调度。我测试过一个极端场景asyncio.Lock在十万次竞争下的开销远低于threading.Lock在同样次数下的开销前提是这些竞争发生在同一个事件循环内。我之前做过一个简单的基准测试创建 10000 个协程去抢一把锁asyncio.Lock整体耗时在毫秒级而用threading.Lock配合线程池模拟同样规模耗时直接到了秒级栈内存占用更是天差地别。1.3 一个类比帮你彻底记住把两种锁类比成不同场景下的排队机制threading.Lock像银行柜台——每个排队的人都是一个实体的线程排在后面的人什么都做不了只能在原地等柜台CPU只能服务当前正在办理的那个人。asyncio.Lock像医院叫号系统——协程就像拿着号的病人没叫到号的时候你可以去喝水、看手机、做检查执行其他协程等叫到号了再过去办理。这个类比虽然不完全精确但对理解为什么异步里不能有阻塞调用非常有帮助如果你在协程里调用threading.Lock的acquire()相当于叫号系统里一个病人站在窗口前干等整个叫号系统被你堵死了。2. 关键API差异context manager、超时与取消语义两种锁虽然都提供了acquire()和release()但在具体 API 语义上有不少细微差别这些差别恰恰是写代码时最容易出问题的地方。2.1 context manager 的真相不只是语法糖两者都支持with或async with语法但内部机制完全不同。threading.Lock实现的是__enter__和__exit__这是 Python 对象协议的一部分with语句会直接调用这两个方法。asyncio.Lock实现的是__aenter__和__aexit__这是异步上下文管理器协议必须配合async with使用。我第一次写异步代码时犯过这样的错误拿着threading.Lock直接在协程函数里写async with lock:结果直接抛出AttributeError因为threading.Lock根本没有__aenter__方法。反过来把asyncio.Lock放在同步函数里写with lock:也会直接报错。这里有一个很容易被忽略的细节asyncio.Lock的acquire()在拿到锁之前不保证公平性也就是说先等待的协程不一定会先获取锁。Python 官方文档里明确提过asyncio.Lock的获取顺序取决于事件循环的调度实际上在 CPython 的早期实现里甚至出现过饥饿问题不过后续版本通过 FIFO 队列做了改进。而threading.Lock在 CPython 中默认也是“非公平”的操作系统调度决定的顺序但因为它本身是阻塞式的这种非公平性对业务的感知不那么明显。2.2 超时参数一个支持一个不支持threading.Lock.acquire(timeout...)是支持的超过指定时间如果还没拿到锁返回False你可以在代码里决定是重试还是放弃import threading lock threading.Lock() if lock.acquire(timeout5.0): try: # 拿到了锁做事情 pass finally: lock.release() else: # 5秒没拿到锁 print(获取锁超时)但asyncio.Lock.acquire()不支持 timeout 参数调用await lock.acquire(timeout5)会直接抛出TypeError。想要超时控制的话需要借助asyncio.wait_for来包装import asyncio lock asyncio.Lock() async def try_acquire(): try: await asyncio.wait_for(lock.acquire(), timeout5.0) except asyncio.TimeoutError: print(获取锁超时处理冲突逻辑) return try: # 拿到了锁做事情 pass finally: lock.release()注意wait_for的超时取消只能取消acquire()的等待过程一旦协程已经拿到了锁wait_for的超时并不会自动释放锁。这点很重要否则会出现锁泄漏——锁永远没人释放其他协程全部卡死。2.3 取消语义协程锁的隐藏陷阱asyncio.Lock还有一个threading.Lock完全不具备的特性它需要考虑任务取消Task Cancellation的影响。当一个协程正在await lock.acquire()时如果外部对这个任务发起了cancel()协程会抛出CancelledError此时锁的等待状态会被正确清理不会污染锁的内部状态。但如果你已经拿到了锁然后业务代码里遇到了异常或取消锁不会自动释放你必须用try/finally来保证释放。这里有个和线程锁类似的经典错误async def bad_example(): async with lock: await some_io_operation() # 如果这里抛异常锁依然会释放 # 但如果锁是在 enter 之前拿到的就必须小心了用async with能保证异常时释放锁这个和with类似。真正容易出问题的是这种写法lock asyncio.Lock() async def risky(): await lock.acquire() # 没有用 async with手动管理 await do_something() lock.release()如果do_something()抛了异常release()永远不会执行锁就泄漏了。所以说实话在绝大多数业务代码里我更推荐直接使用async with lock:这种写法让 Python 帮你兜底。3. 核心区别锁的作用域决定了你该怎么选很多人会忽略一个关键点threading.Lock的锁状态是线程间全局可见的而asyncio.Lock的锁状态只在同一个事件循环内可见。这个区别在最极端的场景下会直接决定程序能不能跑对。3.1 跨线程 vs 跨协程能混用吗直接回答不推荐混用而且大部分情况下混用会出大问题。设想一个场景事件循环中的协程 A 持有了一个threading.Lock同时线程池中的线程 B 也去抢同一把锁。如果 A 和 B 操作的是同一个共享资源这听起来好像“也合理”但问题在于A 拿锁之后如果遇到了真实的 I/O 等待比如await asyncio.sleep()它会主动让出控制权而事件循环在等待期间可能调度其他协程任务如果那些协程任务又试图获取同一把threading.Lock就会发生阻塞事件循环的行为把整个异步循环卡死。反过来如果你在一个协程里await asyncio.Lock但同一个锁还需要被线程池里的同步任务使用这是不能直接用的——asyncio.Lock只能被事件循环内的协程使用你在线程池里调用asyncio.Lock.acquire()不是一个 awaitable 对象在线程里使用的问题而是语义上就不成立。3.2 真实场景选型表场景推荐锁原因纯多线程程序IO 共享变量threading.Lock操作系统线程互斥阻塞等待单线程 异步 I/O 密集任务asyncio.Lock非阻塞协作调度不卡事件循环多个事件循环共享资源推荐使用threading.Lock包裹资源访问或者用外部进程锁asyncio.Lock只对同一循环内协程可见混合场景异步代码调用同步阻塞库用asyncio.to_thread配合threading.Lock保持线程安全避免阻塞事件循环需要跨进程互斥multiprocessing.Lock或第三方库锁进程级互斥线程锁和协程锁都无效这张表是我们在实际项目中总结出来的选型原则核心就一句话锁的作用域必须大于等于共享资源的访问作用域。如果共享资源会被多个线程访问就用线程锁如果只在同一个事件循环里被协程访问就用协程锁。3.3 混合场景的正确姿势最让人头疼的是“异步框架 同步阻塞库”的组合。比如你用httpx的异步客户端没问题但有一天接入了一个只有同步 SDK 的内部服务你不能直接在协程里调用await asyncio.Lock同时又在同步代码里用同一个asyncio.Lock。正确做法是把同步调用丢到线程池用threading.Lock来保护共享状态。我之前做某个内部数据同步项目的时候就是用了这样的模式import asyncio import threading import time import random # 共享状态 shared_cache {} thread_lock threading.Lock() def sync_io_task(key): # 模拟同步阻塞I/O time.sleep(random.uniform(0.1, 0.5)) with thread_lock: value shared_cache.get(key, 0) shared_cache[key] value 1 return shared_cache[key] async def async_worker(key): # 在线程池中执行同步任务 result await asyncio.to_thread(sync_io_task, key) return result async def main(): tasks [async_worker(fkey-{i % 3}) for i in range(30)] results await asyncio.gather(*tasks) print(final shared_cache:, shared_cache) asyncio.run(main())这个模式的实际意义在于把同步阻塞 I/O 放到线程池去执行threading.Lock只保护临界区的共享状态访问不阻塞事件循环。而asyncio.Lock在这种情况下根本用不了因为锁的作用域覆盖不到线程池里的同步代码。4. 实操演示与代码重构从线程版到协程版直接上一段完整的对比代码自己照着跑一遍比看十遍文档都有用。4.1 线程版模拟并发任务交替更新计数器import threading import time counter 0 lock threading.Lock() def worker(): global counter for _ in range(1000): with lock: # 模拟一段短小的临界区操作 temp counter time.sleep(0.0001) counter temp 1 threads [] for _ in range(10): t threading.Thread(targetworker) t.start() threads.append(t) for t in threads: t.join() print(threading counter:, counter)如果你把time.sleep(0.0001)挪出with lock块最终计数结果会小于10000因为多个线程同时读取旧值然后写回丢了更新。这就是典型的竞态条件race condition。4.2 协程版同样的逻辑用 asyncioimport asyncio counter 0 lock asyncio.Lock() async def worker(): global counter for _ in range(1000): async with lock: temp counter # 模拟异步I/O等待但注意锁已经握住了 await asyncio.sleep(0.0001) counter temp 1 async def main(): await asyncio.gather(*(worker() for _ in range(10))) print(asyncio counter:, counter) asyncio.run(main())这两份代码行为几乎完全等价都保证了counter的最终结果为10000。区别在于线程版在临界区里sleep的时候占用的是一条真实的线程资源即使它什么都不做线程被挂起操作系统得调度别的线程。协程版在await asyncio.sleep()的时候事件循环可以继续执行其他协程任务没有“线程被占据”的问题。不过要提醒一句协程版的await asyncio.sleep()放在锁里面等于在临界区里暂停。虽然事件循环不会卡死但锁的持有时间拉长了其他等待锁的协程会排在后面所以临界区的代码应该尽量短小不管是同步还是异步都一样。4.3 从线程版重构到协程版的完整案例现实中的重构往往比上面复杂得多。我在一个模拟场景里处理过一个实际需求原本是 100 个线程并发去调一个第三方 HTTP 接口每个线程访问同一个内存缓存来记录请求计数。后来发现线程开销太大而且第三方接口的响应等待占了大量时间于是想着改成异步方案。初期版本用线程锁import requests import threading cache {} lock threading.Lock() def fetch(url): resp requests.get(url) with lock: cache[url] cache.get(url, 0) len(resp.content)重构为异步版本时很多人会发现requests是同步阻塞库不能直接在协程里用。这时候有两条路用asyncio.to_thread(requests.get, url)把同步请求放到线程池锁仍然用threading.Lock保护缓存。把requests换成httpx.AsyncClient或aiohttp这样全部走异步锁换成asyncio.Lock。推荐方案 2只有彻底异步化才能利用事件循环的优势。但换锁的时候千万别忘了把所有并发入口都改成 async with否则一部分协程用asyncio.Lock另一部分还是threading.Lock共享资源保护就是空的。5. 踩坑实录与排查指南这些坑我基本都踩过一遍这块的实战价值最高。直接把我在实际项目里遇到的几类问题整理出来你对照排查就行。5.1 死锁异步环境下的 串行阻塞先看一个典型的异步死锁场景import asyncio lock asyncio.Lock() async def task_a(): async with lock: await task_b() # 直接调用另一个协程但这个协程也要抢同一种锁 async def task_b(): async with lock: print(never get here) asyncio.run(task_a())执行这个代码程序永远不会打印never get here直接卡死在锁等待上。注意这不是真的死锁而是单事件循环下的锁重入问题——task_a持有锁时调用task_b而task_b尝试获取同一把锁但asyncio.Lock是非可重入的non-reentrant于是永远等待。线程锁也有类似场景一个线程里with lock:再进入同一把lock也会死锁。但异步版本的坑更隐蔽因为await task_b()看起来像函数调用很容易被忽略。排查出这个问题的时候我们在测试环境里直接看到事件循环挂起所有协程都不再输出任何日志一时很难定位。后来靠asyncio.wait_for给所有锁获取加上超时问题才暴露出来——超时异常里直接能看到是哪条协程、哪个文件哪一行等待了哪把锁。5.2 锁泄漏等待方迟迟拿不到锁任务积压asyncio.Lock的锁泄漏通常发生在「手动 acquire 忘记 release」或者「异常路径没有释放」的场景。举个例子lock asyncio.Lock() async def bad_process(): await lock.acquire() try: await call_external_api() # 这里抛异常 finally: pass # 忘记释放锁这个call_external_api()抛异常后finally里没释放锁锁就永远held。后续所有协程在acquire()这里排队时间越久积累的任务越多最终导致内存压力增大甚至进程重启。排查这种问题有个土办法找到锁的关键 waiters 数量做监控。遇到循环调度的 defer 就说明锁泄漏了# 在需要调试的地方插入日志 if lock._waiters: print(flock still has {len(lock._waiters)} waiters)不过我不推荐直接访问私有属性生产环境里去搞可以在锁的 release 处加日志或者用一个简易的计数包装器class LockWithMetrics(asyncio.Lock): def __init__(self): super().__init__() self.acquired_count 0 self.release_count 0 async def acquire(self): await super().acquire() self.acquired_count 1 def release(self): super().release() self.release_count 1监控acquired_count和release_count是否始终平衡如果发散的 lock 增长就说明有泄漏。5.3 性能陷阱临界区里做 I/O 导致吞吐下降可能有人觉得协程锁本身够快于是把大范围的 I/O 操作都放进临界区结果发现整体吞吐急剧下降。原因很简单虽然asyncio.Lock不会阻塞线程但同一把锁在同一时间只允许一个协程进入临界区如果临界区里有一个 300ms 的 HTTP 请求那么每秒最多只能处理约 3 个任务其他请求全部在排队。在这种场景下锁的粒度比锁本身的性能更重要。我的建议是尽量缩短临界区代码只保护真正需要互斥的共享变量读写。把 I/O 操作移出锁外通过“预先取数据、锁内校验/更新”的方式来减少锁持有时间。如果读多写少考虑用asyncio.Semaphore控制并发而不一定用锁互斥。5.4 混用测试ThreadPoolExecutor asyncio.Lock还有一种很隐蔽的坑就是在线程池任务里用了asyncio.Lock。我见过有同事这么写import asyncio from concurrent.futures import ThreadPoolExecutor lock asyncio.Lock() def sync_worker(): # 在线程里尝试用异步锁完全错误 lock.acquire() ...这种写法的结果是acquire()返回了一个 coroutine 对象完全没有执行锁获取当然更没有保护作用。更离谱的是如果你在协程里调用threading.Lock.acquire()虽然能拿到锁但会阻塞事件循环。调试这种 bug 的时候根本找不到具体的抛错信息只有通过性能下降或任务卡死才能看出来。6. 最终经验分享我在实际项目里已经养成了一个习惯在一个项目开始之前先明确“共享资源会被谁访问”——是多个操作系统线程、多个协程还是跨进程。把这个问题的答案写进设计文档再决定使用哪种锁。频繁踩坑之后发现大多数“锁用错了”的问题不是因为不了解锁的 API而是因为对并发边界不够清晰。用asyncio.Lock最核心的心得就是让临界区极短让它内部的 I/O 尽量少用threading.Lock的核心心得则是别在协程里用它也不要在事件循环线程里让它产生真实的阻塞等待。至于哪些代码应该用哪种锁其实只要记住锁的作用域必须覆盖所有可能的访问路径大部分问题都能提前绕开。再补一个小工具层面的建议无论是哪种锁调试期最好都给所有acquire()加上超时机制哪怕是异步锁配合asyncio.wait_for也行。线上没有超时的锁等待是排查死锁时最痛苦的体验加上超时后问题定位会轻松一个数量级。
