Python asyncio 高阶 API 详解：run、gather、wait 与 as_completed 实战

asyncio 高阶 API 概览

在 asyncio 库中，函数通常被划分为低阶和高阶两类。低阶函数主要面向事件循环、Linux 套接字及信号等底层系统功能的直接调用；而高阶函数则封装了更复杂的底层细节，专注于任务并发调度与执行控制。在实际开发场景中，开发者更多使用的是高阶函数。本文将重点介绍 asyncio 中常用的高阶 API 及其用法。

鉴于 asyncio 在不同 Python 版本间存在行为差异，本文及后续系列内容均以 Python 3.10 为基准进行讲解。

run 函数详解

函数原型如下：

asyncio.run(coro, *, debug=False)

功能说明：run() 用于创建新的事件循环，运行传入的协程，并在执行结束后自动关闭该事件循环。它应被视为 asyncio 程序的主入口点。从功能等价性来看，run() 的行为等同于获取一个事件循环，创建一个 Task 对象，将其加入事件循环并运行至完成。

async def main():
    await asyncio.sleep(1)
    print('hello')

asyncio.run(main())

其底层逻辑等价于以下低阶 API 的组合操作：获取事件循环 -> 创建 Task -> 加入循环并运行。

loop = asyncio.get_event_loop()
task = loop.create_task(main())
loop.run_until_complete(task)

create_task 函数详解

函数原型如下：

asyncio.create_task(coro, *, name=None)

功能说明：该函数将协程函数封装为一个 Task 对象。需要注意的是，协程函数本身没有生命周期概念，但 Task 对象具有完整的生命周期。

create_task 会将协程打包为 Task 并自动将其加入当前运行的事件循环，返回 Task 对象。该任务将在 get_running_loop() 返回的循环中执行；若当前线程没有正在运行的事件循环，则抛出 RuntimeError。

async def coro():
    await asyncio.sleep(1)
    print("i am coro")


async def main():
    task = asyncio.create_task(coro())
    print(f"task状态：{task._state}")
    await asyncio.sleep(2)
    print(f"task状态：{task._state}")
    print("i am main")

asyncio.run(main())

执行结果示例：

task状态：PENDING i am coro task 状态：FINISHED i am main

结果分析：

通过输出可见，Task 在运行过程中处于 PENDING 状态，执行完毕后变为 FINISHED 状态，展示了 Task 从创建到完成的生命周期变化。

gather 函数详解

函数原型如下：

asyncio.gather(*aws, return_exceptions=False)

功能说明：

gather 用于并发执行所有传入的可等待对象，收集任务结果并返回一个包含所有已完成 Task 结果的列表。返回值的顺序与传入 Task 的顺序保持一致。可等待对象可以是协程或 Task。

若传入的是协程而非 Task，gather 会自动将其封装为 Task 并加入事件循环。

import asyncio

async def coro(value):
    print(f"hello coro{value}")
    return f"coro{value}"

async def main():
    tasks = [coro(i) for i in range(5)]
    res = await asyncio.gather(*tasks)
    for i in res:
        print(i)

asyncio.run(main())

执行结果示例：

hello coro0 hello coro1 hello coro2 hello coro3 hello coro4 coro0 coro1 coro2 coro3 coro4

结果分析：

该函数成功获取了所有协程的返回值，且返回结果的顺序与任务传入的顺序完全一致。

wait 函数详解

函数原型如下：

asyncio.wait(aws, *, timeout=None, return_when=ALL_COMPLETED)

功能说明：

wait 并发运行序列中的可等待对象，并阻塞直到满足 return_when 指定的条件。它将任务结果收集起来，返回两个 Task/Future 集合：(done, pending)。其中 done 包含已完成的任务，pending 包含未完成的任务（原因可能是超时或特定的 return_when 策略）。

参数 aws：

aws 中保存的是 Task 对象而非协程。从 Python 3.8 版本起，官方不建议传入协程；在 Python 3.11 中，传入协程将不再被支持。

参数 timeout：

若指定 timeout（float 或 int 类型），则该值控制函数返回前等待的最长秒数。

需注意，此函数不会引发 asyncio.TimeoutError。当超时发生时，未完成的 Future 或 Task 将停止执行，且不会返回结果。

参数 return_when：

return_when 指定函数返回的时机，必须为以下参数之一：

基础使用示例：

import asyncio


async def coro(value):
    print(f"hello coro{value}")
    return f"coro{value}"

async def main():
    tasks = [asyncio.create_task(coro(i)) for i in range(5)]
    done, pending = await asyncio.wait(tasks)
    for i in done:
        print(i.result())

asyncio.run(main())

执行结果示例：

hello coro0 hello coro1 hello coro2 hello coro3 hello coro4 coro1 coro2 coro0 coro3 coro4

结果分析：

返回结果的顺序与任务的执行顺序并不一致，体现了并发执行的无序性。

指定超时时间的示例：

import asyncio
from asyncio import FIRST_COMPLETED

async def coro(value):
    print(f"hello coro{value}")
    await asyncio.sleep(value)
    return f"coro{value}"

async def main():
    tasks = [asyncio.create_task(coro(i)) for i in range(5)]
    done, pending = await asyncio.wait(tasks, timeout=3)

    print("---------finish----------")
    for i in done:
        print(i.result())

    print("---------pending----------")
    for i in pending:
        print(i)

asyncio.run(main())

hello coro0 hello coro1 hello coro2 hello coro3 hello coro4 ---------finish---------- coro1 coro2 coro0 ---------pending---------- <Task pending name='Task-5' coro=<coro() running at /Users/ljk/Documents/code/daily_dev/async_demo/wait_demo.py:6> wait_for=<Future finished result=None>> <Task pending name='Task-6' coro=<coro() running at /Users/ljk/Documents/code/daily_dev/async_demo/wait_demo.py:6> wait_for=<Future pending cb=[Task.task_wakeup()]>>

结果分析：

对于超时未完成的 Task，它们会被保存在 pending 集合中。这些未完成的 Task 在超时后不再继续执行，因此没有返回结果。

配置 return_when 为任意任务完成即返回的示例：

import asyncio
from asyncio import FIRST_COMPLETED

async def coro(value):
    print(f"hello coro{value}")
    await asyncio.sleep(value)
    return f"coro{value}"

async def main():
    tasks = [asyncio.create_task(coro(i)) for i in range(5)]
    done, pending = await asyncio.wait(tasks, return_when=FIRST_COMPLETED)

    print("---------finish----------")
    for i in done:
        print(i.result())

    print("---------pending----------")
    for i in pending:
        print(i)

asyncio.run(main())

执行结果示例：

hello coro0 hello coro1 hello coro2 hello coro3 hello coro4 ---------finish---------- coro0 ---------pending---------- <Task pending name='Task-5' coro=<coro() running at /Users/ljk/Documents/code/daily_dev/async_demo/wait_demo.py:6> wait_for=<Future pending cb=[Task.task_wakeup()]>> <Task pending name='Task-3' coro=<coro() running at /Users/ljk/Documents/code/daily_dev/async_demo/wait_demo.py:6> wait_for=<Future pending cb=[Task.task_wakeup()]>> <Task pending name='Task-4' coro=<coro() running at /Users/ljk/Documents/code/daily_dev/async_demo/wait_demo.py:6> wait_for=<Future pending cb=[Task.task_wakeup()]>> <Task pending name='Task-6' coro=<coro() running at /Users/ljk/Documents/code/daily_dev/async_demo/wait_demo.py:6> wait_for=<Future pending cb=[Task.task_wakeup()]>>

结果分析：

当配置为任意任务完成即返回时，一旦获取到一个结果便立即返回，其余未完成的 Task 保存在 pending 中。这些未完成的 Task 在超时后不再继续执行。

as_completed 函数详解

函数原型如下：

asyncio.as_completed(aws, *, timeout=None)

说明：as_completed 并发执行 aws 中保存的可等待对象，返回一个协程的迭代器。可以从该迭代器中取出最先执行完成的 Task 的结果。返回结果的顺序与任务的执行顺序无关。aws 参数可以是 Task 或协程序列。

import asyncio


async def coro(value):
    print(f"hello coro{value}")
    return f"coro{value}"

async def main():
    tasks = [coro(i) for i in range(5)]
    for item in asyncio.as_completed(tasks):
        res = await item
        print(res)


asyncio.run(main())

执行结果示例：

hello coro2 hello coro3 hello coro4 hello coro1 hello coro0 coro2 coro3 coro4 coro1 coro0

结果分析：

所有任务均会执行完成，且未配置超时。返回结果的顺序取决于任务完成的先后，与任务发起顺序无关。

gather、wait、as_completed 异同点小结

在 asyncio 协程体系中，实现创建多个任务并发执行的函数主要有以下三个：

• asyncio.gather
• asyncio.wait
• asyncio.as_completed

不同之处比较：

wait_for 函数详解

函数原型如下：

asyncio.wait_for(aw, timeout)

功能说明：执行单个可等待对象，并指定 timeout 秒数后超时。

wait_for 用于等待可等待对象完成，若指定 timeout 秒数则超时。与 gather 类似，它可以自动将协程转化为 Task 并加入事件循环。

timeout 参数可以为 None，也可以为 float 或 int 型数值表示的等待秒数。如果 timeout 为 None，则一直等待直到完成。若发生超时，任务将被取消并引发 asyncio.TimeoutError。

async def coro():
    # 睡眠5s
    await asyncio.sleep(3600)
    print('finish!')

async def main():
    # Wait for at most 1 second
    try:
        await asyncio.wait_for(coro(), timeout=1.0)
    except asyncio.TimeoutError:
        print('timeout!')

asyncio.run(main())

执行结果示例：

timeout!

以上就是 Python 异步编程中 asyncio 高阶 API 的使用详解，更多关于 Python asyncio 高阶 API 的资料请关注本站其它相关文章！