概述

在 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读取 counter=0
    await asyncio.sleep(0)  # 让出控制权 → 协程B执行
    temp += 1           # 协程A继续，temp=1
    counter = temp      # counter=1（但应为2！）

# 启动两个协程
await asyncio.gather(increment(), increment())
print(counter)  # 可能输出 1（错误！）

✅ 结论：只要存在共享可变状态 + 多个协程修改，就需要同步原语！

1. Lock（锁）

背景和解决的问题

  • 问题：多个协程同时修改共享资源导致数据不一致
  • 解决方案：互斥锁，确保同一时间只有一个协程能访问临界区

使用方法

import asyncio

async def worker(lock, worker_id):
    print(f"Worker {worker_id} waiting for lock")
    async with lock:  # 自动获取和释放锁
        print(f"Worker {worker_id} acquired lock")
        await asyncio.sleep(1)  # 模拟工作
        print(f"Worker {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(f"Worker {worker_id} acquired semaphore")
        await asyncio.sleep(random.uniform(0.5, 2))
        print(f"Worker {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(f"BoundedSemaphore 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(f"Produced: {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(), timeout=2.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(maxsize=10)
    
    # 创建生产者和消费者
    producers = [producer(queue, f"Producer-{i}") for i in range(2)]
    consumers = [consumer(queue, f"Consumer-{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(f"Put: {item} (priority: {priority})")

async def priority_consumer(queue):
    while not queue.empty():
        priority, item = await queue.get()
        print(f"Consumed: {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(f"item-{i}")
    
    # 取出数据（后进先出）
    while not queue.empty():
        item = await queue.get()
        print(f"Got: {item}")
        queue.task_done()

asyncio.run(lifo_example())
# 输出：item-2, item-1, item-0

关键特性

  • 标准的栈行为
  • 适用于需要撤销操作或深度优先处理的场景

9. Timeout（超时）

背景和解决的问题

  • 问题：防止协程无限期等待，提高程序健壮性
  • 解决方案：为异步操作设置时间限制

使用方法

方法1：asyncio.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(), timeout=2.0)
        print(result)
    except asyncio.TimeoutError:
        print("Operation timed out!")

asyncio.run(main())
方法2：asyncio.timeout()（Python 3.11+）
import 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.4+
2. Event二值信号，用于协程间通知无限制❌Python 3.4+
3. Semaphore信号量，控制并发数N个协程❌Python 3.4+
4. BoundedSemaphore有界信号量（防止 release 过度）N个协程❌Python 3.4+
5. Condition条件变量，支持 wait/notify需配合锁❌Python 3.4+
6. QueueFIFO 队列，协程安全无限制✅Python 3.4+
7. PriorityQueue优先级队列（最小堆）无限制✅Python 3.4+
8. LifoQueueLIFO 队列（栈）无限制✅Python 3.4+
9. Timeout上下文管理器式超时控制--Python 3.11+

如何选择？-- 决策指南

asyncio 的同步原语不是“性能杀手”，而是构建正确异步程序的基石。

  1. 简单互斥 → 使用 Lock
  2. 简单通知 → 使用 Event
  3. 资源限流 → 使用 Semaphore 或 BoundedSemaphore
  4. 复杂条件等待 → 使用 Condition
  5. 协程间通信 → 使用各种 Queue
  6. 防止无限等待 → 结合 Timeout 机制

最佳实践

  1. 优先使用上下文管理器：async with lock 比手动 acquire/release 更安全
  2. 避免死锁：注意锁的获取顺序，避免循环等待
  3. 合理设置超时：为所有可能阻塞的操作设置超时
  4. 选择合适的队列类型：根据业务需求选择 FIFO、LIFO 或优先级队列
  5. 及时调用 task_done()：使用 Queue 时不要忘记标记任务完成

这些同步原语是构建健壮异步应用程序的基础工具，正确使用它们可以有效解决并发编程中的各种同步问题。

附：学习建议

  1. 动手写一个异步爬虫，用 Semaphore 限流 + Queue 存结果
  2. 尝试将同步生产者-消费者改造成异步版本
  3. 阅读 aiohttp、FastAPI 源码，看它们如何使用这些原语
# python # 开发语言
Logo
亚马逊云科技技术品牌专区

更多推荐

open-code-review Codex 插件教程：对话式调用评审技能指南

本文是一份面向新手的 **open-code-review Codex 插件教程**，教你在 Codex 中通过对话式方式调用 open-code-review 的 AI 代码评审技能。open-code-review 是阿里巴巴开源的 AI 代码评审工具，采用"确定性流水线 + LLM Agent"混合架构，内置多语言评审规则（NPE、线程安全、XSS、SQL 注入等），可在 Codex 里用一

avatar 亚马逊云科技技术品牌专区

OBS Studio 完整指南：免费开源录屏与直播软件从零上手

想做一场像模像样的直播，或者把屏幕内容录成教程视频，但一想到商用软件的订阅费就有些犹豫？其实你不需要花一分钱。OBS Studio 是一款免费开源的直播与录屏软件，支持 Windows、macOS 和 Linux，采集、混音、编码、推流、录制，一条流水线全包了。## 先搞清楚 OBS Studio 是什么OBS Studio 是一个"场景化"的采集合成工具：你把摄像头、窗口、浏览器、图片、

avatar 亚马逊云科技技术品牌专区

如何用MediaCrawler爬取小红书抖音数据？7大平台爬虫从安装到实战完整指南

MediaCrawler 是一个支持小红书、抖音、快手、B站、微博、百度贴吧、知乎的开源自媒体数据采集工具。它能按关键词、按帖子、按创作者主页三种方式抓取笔记/视频内容、评论甚至二级评论，并把结果存成 JSON、CSV、Excel 或数据库。适合做市场调研、竞品监控、内容选题的运营人员，以及想要现成爬虫框架而不想从零写 JS 逆向的开发者。## 从一次"手动抄数据"说起想象这个场景：你负责

avatar 亚马逊云科技技术品牌专区