概述在 Python 异步编程中虽然协程coroutine天然避免了线程切换开销但多个协程仍可能同时访问共享资源如全局变量、文件、数据库连接从而引发竞态条件Race Condition。为此asyncio提供了一套专为异步环境设计的同步原语Synchronization Primitives包括Lock、Semaphore、Event、Queue等。本文将深入浅出地介绍这些原语的原理、用法、区别及典型应用场景助你写出安全、高效的异步代码。重要提示asyncio 的同步原语与 threading 模块的对应原语类似但专为异步环境设计不能在同步代码中使用也不能与 threading 原语混用。0. 为什么异步也需要“锁”很多人误以为“异步 单线程 不需要锁”。这是错误的虽然asyncio默认在单线程运行但协程会在await处主动让出控制权。如果两个协程同时操作共享变量就可能发生交错执行# 危险示例无锁累加 counter 0 async def increment(): global counter temp counter # 协程A读取 counter0 await asyncio.sleep(0) # 让出控制权 → 协程B执行 temp 1 # 协程A继续temp1 counter temp # counter1但应为2 # 启动两个协程 await asyncio.gather(increment(), increment()) print(counter) # 可能输出 1错误✅结论只要存在共享可变状态 多个协程修改就需要同步原语1. Lock锁背景和解决的问题问题多个协程同时修改共享资源导致数据不一致解决方案互斥锁确保同一时间只有一个协程能访问临界区使用方法import asyncio async def worker(lock, worker_id): print(fWorker {worker_id} waiting for lock) async with lock: # 自动获取和释放锁 print(fWorker {worker_id} acquired lock) await asyncio.sleep(1) # 模拟工作 print(fWorker {worker_id} releasing lock) async def main(): lock asyncio.Lock() tasks [worker(lock, i) for i in range(3)] await asyncio.gather(*tasks) asyncio.run(main())关键特性支持async with语法推荐也可以手动调用await lock.acquire()和lock.release()不可重入同一个协程不能多次获取同一个锁性能开销极小纯用户态切换。2. Event事件背景和解决的问题问题一个或多个协程需要等待某个条件发生解决方案提供信号机制允许协程等待和通知使用方法import asyncio async def waiter(event, name): print(f{name} waiting for event) await event.wait() # 阻塞直到事件被设置 print(f{name} received event) async def setter(event): await asyncio.sleep(2) print(Setting event) event.set() # 设置事件唤醒所有等待者 async def main(): event asyncio.Event() tasks [ waiter(event, Worker 1), waiter(event, Worker 2), setter(event) ] await asyncio.gather(*tasks) asyncio.run(main())涉及的方法event.wait()等待事件被设置event.set()设置事件唤醒所有等待协程event.clear()清除事件状态event.is_set()检查事件是否已设置关键特性信号持久化set()后后续wait()立即返回适合一次性启动/就绪通知3. Semaphore信号量背景和解决的问题问题限制同时访问某个资源的协程数量如数据库连接池、API 调用限制解决方案计数信号量允许多个但有限数量的协程同时访问使用方法import asyncio import random async def worker(semaphore, worker_id): async with semaphore: # 获取信号量 print(fWorker {worker_id} acquired semaphore) await asyncio.sleep(random.uniform(0.5, 2)) print(fWorker {worker_id} released semaphore) async def main(): # 最多允许2个协程同时执行 semaphore asyncio.Semaphore(2) tasks [worker(semaphore, i) for i in range(5)] await asyncio.gather(*tasks) asyncio.run(main())关键特性初始化时指定最大并发数支持async with语法可以动态调整通过acquire()/release()Semaphore(1)等价于Lock常用于连接池、爬虫限速、数据库并发控制4. BoundedSemaphore有界信号量背景和解决的问题问题防止信号量的 release() 调用次数超过 acquire()避免逻辑错误解决方案在普通信号量基础上增加边界检查使用方法import asyncio async def example(): bounded_sem asyncio.BoundedSemaphore(2) # 正常使用 await bounded_sem.acquire() await bounded_sem.acquire() # 这会抛出 ValueError因为已经达到了初始值 try: bounded_sem.release() bounded_sem.release() bounded_sem.release() # 这里会报错 except ValueError as e: print(fBoundedSemaphore error: {e}) asyncio.run(example())与 Semaphore 的区别功能完全相同只是增加了安全检查如果release()调用次数超过初始值会抛出ValueError推荐在不确定是否会过度释放的场景使用5. Condition条件变量背景和解决的问题问题协程需要等待某个复杂条件成立而不仅仅是简单的信号解决方案结合锁和事件支持条件等待和通知使用方法import asyncio async def consumer(condition, name): async with condition: print(f{name} waiting for item) await condition.wait() # 等待条件满足 print(f{name} consumed item) async def producer(condition): await asyncio.sleep(1) async with condition: print(Producing item) condition.notify_all() # 通知所有等待者 async def main(): condition asyncio.Condition() tasks [ consumer(condition, Consumer 1), consumer(condition, Consumer 2), producer(condition) ] await asyncio.gather(*tasks) asyncio.run(main())关键特性必须在持有锁的情况下调用wait()、notify()、notify_all()wait()会释放锁并阻塞被唤醒后重新获取锁notify(n)唤醒 n 个等待者notify_all()唤醒所有等待者⚠️ 重要必须用while检查条件防止虚假唤醒底层基于Lock自动管理锁6. Queue队列背景和解决的问题问题协程间安全地传递数据实现生产者-消费者模式解决方案线程安全的异步队列使用方法import asyncio import random async def producer(queue, name): for i in range(5): item f{name}-item-{i} await queue.put(item) print(fProduced: {item}) await asyncio.sleep(random.uniform(0.1, 0.5)) async def consumer(queue, name): while True: try: item await asyncio.wait_for(queue.get(), timeout2.0) print(f{name} consumed: {item}) queue.task_done() await asyncio.sleep(random.uniform(0.1, 0.3)) except asyncio.TimeoutError: break async def main(): queue asyncio.Queue(maxsize10) # 创建生产者和消费者 producers [producer(queue, fProducer-{i}) for i in range(2)] consumers [consumer(queue, fConsumer-{i}) for i in range(3)] # 等待所有生产者完成 await asyncio.gather(*producers) # 等待队列中所有任务完成 await queue.join() # 取消消费者任务 for c in consumers: c.cancel() asyncio.run(main())关键特性put(item)放入项目如果队列满则阻塞get()取出项目如果队列空则阻塞task_done()标记任务完成join()等待所有任务完成支持最大容量限制7. PriorityQueue优先级队列背景和解决的问题问题需要按优先级处理任务解决方案基于优先级的队列优先级低的先出队使用方法import asyncio async def priority_producer(queue): # 元组格式(priority, item) items [(3, low priority), (1, high priority), (2, medium priority)] for priority, item in items: await queue.put((priority, item)) print(fPut: {item} (priority: {priority})) async def priority_consumer(queue): while not queue.empty(): priority, item await queue.get() print(fConsumed: {item} (priority: {priority})) queue.task_done() async def main(): queue asyncio.PriorityQueue() await priority_producer(queue) await priority_consumer(queue) asyncio.run(main())关键特性基于 heapq 实现优先级数值越小优先级越高项目必须是可比较的8. LifoQueue后进先出队列背景和解决的问题问题需要栈式LIFO的数据处理顺序解决方案后进先出的队列使用方法import asyncio async def lifo_example(): queue asyncio.LifoQueue() # 放入数据 for i in range(3): await queue.put(fitem-{i}) # 取出数据后进先出 while not queue.empty(): item await queue.get() print(fGot: {item}) queue.task_done() asyncio.run(lifo_example()) # 输出item-2, item-1, item-0关键特性标准的栈行为适用于需要撤销操作或深度优先处理的场景9. Timeout超时背景和解决的问题问题防止协程无限期等待提高程序健壮性解决方案为异步操作设置时间限制使用方法方法1asyncio.wait_for()import asyncio async def slow_operation(): await asyncio.sleep(5) return Done async def main(): try: result await asyncio.wait_for(slow_operation(), timeout2.0) print(result) except asyncio.TimeoutError: print(Operation timed out!) asyncio.run(main())方法2asyncio.timeout()Python 3.11import asyncio async def main(): try: async with asyncio.timeout(2.0): await asyncio.sleep(5) except TimeoutError: print(Timed out!) asyncio.run(main())方法3手动实现超时import asyncio async def with_timeout(coro, timeout): try: return await asyncio.wait_for(coro, timeout) except asyncio.TimeoutError: raise TimeoutError(Operation timed out) # 使用 # result await with_timeout(some_async_func(), 5.0)关键特性超时后自动取消任务是构建健壮异步系统的必备工具各原语对比 总结原语说明并发控制数据传递引入版本1.Lock互斥锁保护临界区1个协程❌Python 3.42.Event二值信号用于协程间通知无限制❌Python 3.43.Semaphore信号量控制并发数N个协程❌Python 3.44.BoundedSemaphore有界信号量防止 release 过度N个协程❌Python 3.45.Condition条件变量支持 wait/notify需配合锁❌Python 3.46.QueueFIFO 队列协程安全无限制✅Python 3.47.PriorityQueue优先级队列最小堆无限制✅Python 3.48.LifoQueueLIFO 队列栈无限制✅Python 3.49.Timeout上下文管理器式超时控制--Python 3.11如何选择-- 决策指南asyncio的同步原语不是“性能杀手”而是构建正确异步程序的基石。简单互斥→ 使用Lock简单通知→ 使用Event资源限流→ 使用Semaphore或BoundedSemaphore复杂条件等待→ 使用Condition协程间通信→ 使用各种Queue防止无限等待→ 结合Timeout机制最佳实践优先使用上下文管理器async with lock比手动 acquire/release 更安全避免死锁注意锁的获取顺序避免循环等待合理设置超时为所有可能阻塞的操作设置超时选择合适的队列类型根据业务需求选择 FIFO、LIFO 或优先级队列及时调用 task_done()使用 Queue 时不要忘记标记任务完成这些同步原语是构建健壮异步应用程序的基础工具正确使用它们可以有效解决并发编程中的各种同步问题。附学习建议动手写一个异步爬虫用Semaphore限流 Queue存结果尝试将同步生产者-消费者改造成异步版本阅读aiohttp、FastAPI源码看它们如何使用这些原语