没吃饱的笨蛋

外观

asyncio 同步原语

约 1290 字大约 4 分钟

2026-05-10

阅读导览
asyncio 同步原语

虽然 asyncio 默认运行在单线程中，但协程之间仍然可能出现共享状态问题。只要在读取和写入之间存在 await，其他协程就可能插入执行。

  • 协程中的竞态条件
  • asyncio.Lock
  • asyncio.Event
  • asyncio.Queue
实践清单
学习练习清单
  1. 1跑 1000 个 async def add_one() 看 counter 不是 1000 —— 协程在 await 处会切换，单线程也有竞态。
  2. 2用 async with lock: 包住共享状态读写区，但锁住的范围越小越好；不要在锁里 await 长 IO。
  3. 3asyncio.Semaphore(5) 限制 100 个 URL 请求最多同时 5 个 —— 是控外部限流的标准写法。
  4. 4不要拿 threading.Lock 给协程加锁；asyncio 有自己的 Lock/Event/Queue/Semaphore。
  5. 5能用 asyncio.Queue 走生产者消费者，就别上来堆共享变量加锁。

虽然 asyncio 默认运行在单线程中，但协程之间仍然可能出现共享状态问题。只要在读取和写入之间存在 await，其他协程就可能插入执行。

协程中的竞态条件

import asyncio

counter = 0

async def add():
    global counter
    value = counter
    await asyncio.sleep(0)
    counter = value + 1

多个协程同时执行时，可能读到相同旧值，然后互相覆盖结果。

asyncio.Lock

import asyncio

counter = 0
lock = asyncio.Lock()

async def add():
    global counter
    async with lock:
        value = counter
        await asyncio.sleep(0)
        counter = value + 1

asyncio.Lock 要配合 async with 使用。

注意：不要用 threading.Lock 来保护协程代码。asyncio 有自己的同步原语。

asyncio.Event

Event 用来让一个或多个协程等待某个信号。

import asyncio

event = asyncio.Event()

async def waiter():
    print('等待事件')
    await event.wait()
    print('事件已发生')

async def setter():
    await asyncio.sleep(1)
    event.set()

async def main():
    await asyncio.gather(waiter(), setter())

asyncio.run(main())

asyncio.Queue

asyncio.Queue 用来在协程之间传递数据，适合异步生产者消费者模型。

import asyncio

async def producer(q):
    for i in range(5):
        await q.put(i)
        print(f'生产 {i}')

async def consumer(q):
    while True:
        item = await q.get()
        try:
            print(f'消费 {item}')
        finally:
            q.task_done()

async def main():
    q = asyncio.Queue()
    consumer_task = asyncio.create_task(consumer(q))
    await producer(q)
    await q.join()
    consumer_task.cancel()

asyncio.run(main())

Semaphore

Semaphore 用来限制同时进入某段代码的协程数量。

semaphore = asyncio.Semaphore(5)

async def fetch(url):
    async with semaphore:
        await asyncio.sleep(1)
        return url

适合限制并发请求、数据库连接、文件操作等。

Condition

asyncio.Condition 可以让协程等待更复杂的状态变化。实际项目中使用频率不如 Lock、Event、Queue 高。

condition = asyncio.Condition()

一般初学阶段先掌握 Lock、Event、Queue、Semaphore 即可。

注意事项

  1. 单线程协程不等于没有并发问题。
  2. 只要在状态更新中间有 await，就可能产生竞态条件。
  3. asyncio 同步原语不能跨线程随意使用。
  4. 优先用 Queue 传递数据，减少共享状态。
  5. 不要在协程中使用阻塞式同步锁。

单线程 asyncio 也会有竞态条件

很多人以为 asyncio 是单线程，所以不会有并发问题。其实只要多个协程共享状态，并且中间有 await，就可能出现竞态条件。

import asyncio

counter = 0


async def add_one():
    global counter
    value = counter
    await asyncio.sleep(0)
    counter = value + 1


async def main():
    await asyncio.gather(*(add_one() for _ in range(1000)))
    print(counter)


asyncio.run(main())

你可能期待输出 1000，但结果可能不是。因为协程在 await 的地方会让出控制权。

asyncio.Lock 保护共享状态

import asyncio

counter = 0
lock = asyncio.Lock()


async def add_one():
    global counter
    async with lock:
        counter += 1

和线程锁一样，锁住的范围要小。不要在锁里做慢请求或长时间等待，否则其他协程都会被挡住。

Semaphore 控制并发数量

假设你要请求 100 个 URL，但最多同时请求 5 个：

import asyncio

sem = asyncio.Semaphore(5)


async def fetch(url):
    async with sem:
        print('请求', url)
        await asyncio.sleep(1)
        return url

Semaphore 很适合限制接口请求、文件处理、数据库连接等资源使用量。

Queue 组织生产者消费者

import asyncio


async def producer(q):
    for i in range(5):
        await q.put(i)
    await q.put(None)


async def consumer(q):
    while True:
        item = await q.get()
        try:
            if item is None:
                return
            print('处理', item)
        finally:
            q.task_done()


async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))


asyncio.run(main())

asyncio.Queue 和线程里的 queue.Queue 思路很像，只是操作前面要加 await。

初学者常用同步工具怎么选

工具解决什么问题
Lock保护共享状态，避免同时修改
Event一个协程通知其他协程“某件事发生了”
Queue在协程之间传递任务或数据
Semaphore限制同时运行的任务数量
Condition更复杂的条件等待，初学阶段少用

如果能用 Queue 传消息，就不要急着共享一堆变量再加锁。

总结

asyncio 的同步原语用于协调协程之间的执行关系。它们和 threading 中的工具功能类似，但使用方式是异步的。写 asyncio 程序时，仍然要认真处理共享状态、任务通知和并发限制。

重点总结
要点回收
  • 单线程 asyncio 也可能有竞态条件，因为协程会在 `await` 处切换。
  • `asyncio.Lock` 保护共享状态，`Semaphore` 控制并发数量，`Queue` 传递任务。
  • 能用 Queue 组织数据流时，通常比共享变量加锁更清晰。

版权归属：Shuo Liu

许可证：署名-非商业性 4.0 国际 (CC-BY-NC-4.0)