跳至内容

asyncio 异步

asyncio 的基本用法与常见问题。

发布于 更新于 约 11 分钟阅读 浏览 — 次
  • Python
  • asyncio
  • 协程
  • 异步
  • 异步编程
  • 队列

asyncio 通过事件循环调度任务，让多个 I/O 操作的等待时间重叠。

运行环境：Python 3.13，默认启用 GIL 的 CPython。

1. 同步与异步

这里比较同步阻塞调用与 asyncio 异步调用：

对比项同步调用asyncio 异步调用
等待 I/O调用线程等待操作完成当前任务挂起，事件循环可以运行其他任务
取得结果函数调用结束后返回用 await 等待结果
多项操作并发通常借助线程池等机制把多个协程调度为 Task
使用条件库提供同步接口库提供异步接口，调用链能配合 await
编写与排查调用顺序直观需要处理任务之间的等待、取消和共享资源

优点：多个 I/O 操作可以重叠等待，一个线程就能管理多个任务，减少为每个等待中的操作分配线程的开销。

限制：阻塞调用或长时间计算会占住事件循环线程，影响其他任务；依赖的库也需要支持异步接口。纯 Python 重计算不会因此自动使用多个 CPU 核心。

异步不等于并发。逐个 await 协程仍可按顺序执行；任务少、等待短或前后依赖紧密时，改成异步未必有收益。

2. 协程与事件循环

2.1 协程函数与协程对象

用 async def 定义协程函数。调用它时，先创建一个协程对象；此时函数体还没有开始执行。

示例：

代码块 Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import asyncio


async def fetch_price(product_id):
    print(f"开始查询 {product_id}")
    return {"product_id": product_id, "price": product_id * 10}


async def main():
    query = fetch_price(101)
    print("已创建协程对象")
    result = await query
    print(result)


asyncio.run(main())
运行结果 Text
1
2
3

# 已创建协程对象
# 开始查询 101
# {'product_id': 101, 'price': 1010}
名称类型说明
fetch_price协程函数定义查询逻辑
query协程对象由 fetch_price(101) 创建，此时查询尚未执行
result字典await query 完成后取得的查询结果

只创建协程，却没有等待它或交给任务调度，函数体就不会执行。协程对象被回收时会出现以下警告，这里省略文件路径和行号：

运行结果 Text
1

# RuntimeWarning: coroutine 'fetch_price' was never awaited

2.2 await 与事件循环

可等待对象是能放在 await 后面的对象，常见的有协程对象、Task 和 asyncio.Future。普通字典、整数不能直接 await。

await query 直接等待协程。也可以先用 create_task() 把协程包装为 Task（任务对象），交给事件循环调度。把上例的 main() 换成：

代码块 Python
1
2
3
4

async def main():
    task = asyncio.create_task(fetch_price(101))
    result = await task
    print(result)
运行结果 Text
1
2

# 开始查询 101
# {'product_id': 101, 'price': 1010}

create_task() 返回 Task，本身不等待任务完成。await task 在任务尚未结束时挂起当前任务，完成后取得返回值。

同一事件循环线程中，同一时刻只执行一个任务的代码。 一个任务挂起后，事件循环可以运行其他就绪任务；没有任务就绪时，可以等待 I/O 或定时器通知。

await 不保证切换任务。 如果协程能直接返回，或所等待的 Task 已经完成，当前任务可以接着执行。长时间计算即使用 async def 定义，也不会自动让出执行机会。

2.3 创建与运行事件循环

事件循环负责调度任务、执行回调，以及等待 I/O 和定时器通知。脚本通常通过 asyncio.run(main()) 启动：它创建循环，运行 main()，结束时清理资源并关闭循环。

接口作用
asyncio.run(main())创建、运行并关闭事件循环，返回 main() 的结果
asyncio.get_running_loop()获取当前线程正在运行的循环；没有时抛出 RuntimeError
asyncio.get_event_loop()获取当前线程的当前循环；在协程内返回正在运行的循环
asyncio.new_event_loop()创建循环对象，不启动它
asyncio.set_event_loop(loop)将循环设为当前线程的当前循环，不启动它

协程内部需要获取循环时，优先用 get_running_loop()。

需要手动管理时，可以把创建、运行和关闭分开：

代码块 Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

import asyncio


async def main():
    loop = asyncio.get_running_loop()
    print(loop.is_running())
    return 1010


loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
    print(asyncio.get_event_loop() is loop)
    result = loop.run_until_complete(main())
    print(result)
finally:
    loop.close()
    asyncio.set_event_loop(None)
运行结果 Text
1
2
3

# True
# True
# 1010

run_until_complete() 驱动循环，直到传入的协程结束。close() 关闭循环，但不会替你等待未结束的任务。普通脚本使用 asyncio.run() 即可，手动管理还需要自行处理剩余任务、异步生成器和线程池的清理。

Jupyter 等已有事件循环的环境通常直接写 await main()。同一线程已有事件循环在运行时，不能再调用 asyncio.run()。

3. 顺序执行与并发执行

3.1 create_task()

sleep() 与 create_task() 作用不同：

  • await asyncio.sleep(delay) 暂停当前任务。
  • create_task() 创建并调度一个任务。创建 Task 不会让其中的阻塞代码变成异步代码。
代码块 Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

import asyncio


async def fetch_price(product_id, delay):
    print(f"开始 {product_id}")
    await asyncio.sleep(delay)
    print(f"完成 {product_id}")
    return product_id * 10


async def main():
    print("顺序等待")
    first = await fetch_price(101, 0.2)
    second = await fetch_price(102, 0.1)
    print([first, second])

    print("并发执行")
    first_task = asyncio.create_task(fetch_price(101, 0.2))
    second_task = asyncio.create_task(fetch_price(102, 0.1))
    first = await first_task
    second = await second_task
    print([first, second])


asyncio.run(main())

使用默认任务调度方式时，示例输出如下：

运行结果 Text
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# 顺序等待
# 开始 101
# 完成 101
# 开始 102
# 完成 102
# [1010, 1020]
# 并发执行
# 开始 101
# 开始 102
# 完成 102
# 完成 101
# [1010, 1020]

顺序写法先等 101 返回，再调用 102。并发写法通过 create_task() 创建两个 Task（任务对象），把它们交给事件循环调度，然后等待结果。

并发部分的执行过程如下：

步骤执行过程等待中的查询
创建任务main() 创建两个 Task，随后等待 101两个查询尚未开始
启动查询101 打印开始，进入等待；102 打印开始，进入等待101、102
102 的等待结束恢复 102，打印完成并返回价格101
101 的等待结束恢复 101，打印完成并返回价格无
取齐结果main() 拿到 101 的价格，再取得已完成的 102 的价格无

按顺序等待 Task，不会把已经调度的任务变成串行。 await first_task 暂停的是 main()，102 仍然可以继续执行。

3.2 任务结果

  • await task：等待任务，成功时返回结果，失败时抛出异常。
  • task.result()：直接读取结果，不会等待；任务尚未完成时会报错。

保留 create_task() 返回的 Task，方便等待结果、处理异常；一组任务可以交给 TaskGroup 管理。

4. TaskGroup 与结果收集

4.1 批量任务

TaskGroup 创建并管理一组任务，退出时等待组内任务结束。

async with 是异步上下文管理语法，允许在进入、退出时执行异步操作。任务组的等待发生在退出时：

代码块 Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

import asyncio


async def fetch_price(product_id):
    await asyncio.sleep(0.01)
    return {"product_id": product_id, "price": product_id * 10}


async def main():
    product_ids = [101, 102, 103]
    async with asyncio.TaskGroup() as group:
        tasks = {product_id: group.create_task(fetch_price(product_id)) for product_id in product_ids}

    results = {product_id: task.result() for product_id, task in tasks.items()}
    print(results[102])


asyncio.run(main())
运行结果 Text
1

# {'product_id': 102, 'price': 1020}

所有任务成功、正常离开 async with 后，才读取 task.result()。

4.2 异常处理

任务组中的任务抛出非 CancelledError 异常时，会触发以下处理：

  1. 请求取消组内其余未完成任务。
  2. 等待这些任务清理并退出。
  3. 将普通异常合并成 ExceptionGroup 抛出。

下面让 102 查询失败，101 仍处于等待中：

代码块 Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

import asyncio


async def fetch_price(product_id):
    try:
        if product_id == 102:
            await asyncio.sleep(0.01)
            raise ConnectionError("102 价格服务不可用")
        await asyncio.sleep(1)
        return product_id * 10
    finally:
        print(f"结束查询 {product_id}")


async def main():
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(fetch_price(101))
            group.create_task(fetch_price(102))
    except* ConnectionError as errors:
        print([str(error) for error in errors.exceptions])


asyncio.run(main())
运行结果 Text
1
2
3

# 结束查询 102
# 结束查询 101
# ['102 价格服务不可用']

102 抛出异常后先执行 finally。101 随后被取消，也执行 finally，但没有返回价格。两条“结束查询”都只是清理阶段的输出，不表示查询成功。

except* ConnectionError 从异常组中选出匹配的异常，组成异常组交给 errors。本例用 errors.exceptions 取得其中的连接错误并打印；如果组内还有未被处理的其他异常，它们会继续向外抛出。

4.3 create_task、TaskGroup 与 gather

  • asyncio.create_task() 创建单个任务；

  • TaskGroup 管理一组任务的等待和退出；

  • gather() 汇总多个操作的结果；

  • group.create_task() 会把新任务登记到任务组，普通的 asyncio.create_task() 不会。

接口结果与等待普通异常的处理
asyncio.create_task(coro)返回单个 Task，需要自行等待或读取结果等待或读取结果时抛出；不自动取消其他独立任务
group.create_task(coro)返回单个 Task，退出 TaskGroup 时统一等待任务组取消其余任务并等待退出，再报告异常组
asyncio.gather(...)等待后得到列表，顺序对应传入顺序默认立即抛出第一个异常，不自动取消其他任务

gather() 可以接收已经创建的 Task；传入协程时，它会自动把协程调度为 Task。

默认的 gather() 不会因为一项失败就取消其余任务。但如果异常使 main() 退出，asyncio.run() 会清理剩余任务；这是运行入口的收尾行为。

4.4 按完成顺序取结果

asyncio.as_completed() 适合每完成一项就保存结果或更新进度。

Python 3.13 起，可以用 async for 遍历它。传入 Task 时，异步迭代返回已完成的原 Task，便于关联输入。async for 可以等待下一项，但不会让循环体自动并发。

5. 超时与取消

5.1 单次调用超时

asyncio.timeout() 限制一段异步操作的等待时间。示例设置 0.02 秒超时，查询需要等待 1 秒：

代码块 Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

import asyncio


async def fetch_price(product_id):
    try:
        await asyncio.sleep(1)
        return product_id * 10
    finally:
        print(f"结束查询 {product_id}")


async def main():
    try:
        async with asyncio.timeout(0.02):
            price = await fetch_price(101)
            print(price)
    except TimeoutError:
        print("本次查询超时")


asyncio.run(main())
运行结果 Text
1
2

# 结束查询 101
# 本次查询超时

超时后，timeout() 请求取消当前任务。fetch_price() 收到取消异常，执行 finally 后退出。

离开超时上下文时，这次取消会被转换成 TimeoutError。因此，except TimeoutError 要写在 async with 外侧。

5.2 整批任务超时

把任务组放在同一个 timeout() 内，可以让整批查询共用 2 秒：

代码块 Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

import asyncio


async def fetch_price(product_id):
    await asyncio.sleep(0.01)
    return {"product_id": product_id, "price": product_id * 10}


async def main():
    try:
        async with asyncio.timeout(2):
            async with asyncio.TaskGroup() as group:
                tasks = [group.create_task(fetch_price(product_id)) for product_id in [101, 102, 103]]
        print([task.result()["price"] for task in tasks])
    except TimeoutError:
        print("整批查询超时")


asyncio.run(main())
运行结果 Text
1

# [1010, 1020, 1030]

整批超时从进入外层 timeout() 开始计算。给每个查询分别设置 2 秒，则是各自计时，不能保证整批在 2 秒内结束。timeout() 从 Python 3.11 起提供。

5.3 取消与清理

task.cancel() 发出取消请求。任务在后续可处理取消的位置收到 CancelledError，不会从任意一行被强行终止。

关闭文件、释放连接等清理操作放在 finally 中，或交给资源的上下文管理器处理。这样正常结束、异常和取消都能执行清理。

捕获 CancelledError 后，通常应在清理完毕后重新 raise。 任务组和超时管理依赖取消来结束工作，吞掉它可能让外层继续等待。

标准库中的定义如下：

代码块 Python
1
2

class CancelledError(BaseException):
    """The Future or Task was cancelled."""

它直接继承 BaseException，因此 except Exception 不会捕获它；需要单独处理时写 except asyncio.CancelledError。

超时不保证立即返回。 阻塞调用会使事件循环无法及时处理取消，清理操作本身也可能耗时。取消也不会撤销已经完成的远端写入。

单个可等待操作也可以用 asyncio.wait_for(awaitable, timeout=...) 设置超时。超时后，它会请求取消并等待取消处理，所以实际耗时也可能超过指定值。

6. 并发控制

6.1 Semaphore

需要给一组异步操作设置并发上限时，可以使用标准库的 asyncio.Semaphore。它适合限制同时占用某项资源的任务数，不是每段异步代码都需要加。

Semaphore 保存可用名额数。进入 async with semaphore 时取得一个名额，退出时归还；名额用完，后来的任务就等待。

代码块 Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

import asyncio


async def fetch_price(product_id, semaphore):
    async with semaphore:
        await asyncio.sleep(0.01)
        return product_id * 10


async def main():
    semaphore = asyncio.Semaphore(2)
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(fetch_price(product_id, semaphore)) for product_id in range(101, 107)]
    print([task.result() for task in tasks])


asyncio.run(main())
运行结果 Text
1

# [1010, 1020, 1030, 1040, 1050, 1060]

六个任务共用同一个 semaphore，最多两个进入查询区间，其余任务等待名额。每个任务各建一个信号量，无法限制整批并发。

工程中也可能由连接池、任务队列或框架控制容量。已有组件的限制能覆盖同一资源时，可以直接使用其配置；需要跨多个操作限制业务并发时，再共享一个信号量。这个上限只对共享它的任务生效，多个进程或服务实例之间不会自动合并计数。

6.2 并发数、请求速率与任务数量

限制项作用常用方法
同时进行的请求数下游同时承受的工作量共享的 Semaphore、客户端连接池
每秒请求数单位时间发起多少请求速率限制策略
已创建、尚未结束的任务数内存占用和排队规模分批创建任务，或使用有界队列和固定数量的处理任务

并发上限不等于每秒请求数。 请求返回越快，名额释放越快，后续请求就能更早开始。连接数也不等于请求并发数：HTTP/2 可以在一个连接上承载多个请求。

信号量不限制 Task 的创建数量。 即使同时只允许 10 个任务查询，一次创建的大量 Task 仍会占用内存、等待名额。

数据量大时，可以分批创建任务，处理完一批再取下一批。这样会等该批最慢的任务结束；需要持续补充任务时，可以使用有界队列和固定数量的处理任务。

多个入口各自创建 Semaphore(10)，合计并发仍可能超过 10。要限制这些入口的总并发，需要让它们在同一事件循环内共享一个信号量。

7. 调用同步函数

如果现有 SDK 只提供同步函数，可以用 asyncio.to_thread() 把调用交给线程执行，避免它占住事件循环线程。

代码块 Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

import asyncio
from time import sleep


def fetch_price_sync(product_id):
    sleep(0.05)
    return product_id * 10


async def main(item):
    return await asyncio.to_thread(fetch_price_sync, item)


print(asyncio.run(main(1010)))
运行结果 Text
1

# 10100

await 取得结果，return 将结果传给调用方。少了 return，main() 默认返回 None，asyncio.run() 得到的也就是 None。

传入函数和参数，不要先调用函数。 to_thread(fetch_price_sync, item) 把调用交给线程；写成 to_thread(fetch_price_sync(item)) 则会先在当前线程执行同步函数，仍然阻塞事件循环。

to_thread() 返回协程，需要 await 或交给 Task 调度。线程执行同步函数时，事件循环线程可以继续处理其他任务。若多个线程共用 SDK 客户端，还要确认它是否支持线程安全访问。

取消等待，不能强行停止已经运行的线程函数。 SDK 本身仍需要超时设置。asyncio.run() 退出时也要处理线程池关闭，取消等待不等于底层调用已经结束。

默认启用 GIL 的 CPython 中，to_thread() 通常用于阻塞 I/O。纯 Python 重计算通常考虑进程池；会释放 GIL 的扩展计算则按具体库判断。

评论

使用 GitHub 登录后参与讨论，评论将公开保存在 GitHub。

阅读到这里时加载评论。