asyncio 任务管理

约 1403 字大约 5 分钟

2026-05-10

asyncio 任务管理

掌握 async def 和 await 后，还需要学习如何管理多个协程任务。asyncio 提供了 Task、gather、wait、ascompleted、超时和取消等工具。

createtask
gather
ascompleted
wait

学习练习清单

1对比 await work('A'); await work('B') 和 asyncio.gather(...)：前者顺序、后者并发。
2gather 按入参顺序返回结果；用 as_completed 谁先完成先处理 —— 爬虫批量请求常用。
3用 asyncio.wait_for(task, timeout=3) 给慢任务设上限，超时自动抛 TimeoutError。
4对长任务调 task.cancel()，在协程里捕获 asyncio.CancelledError 做清理后再 raise。
5gather(return_exceptions=True) 把异常作为结果返回 —— 否则一个任务失败会拖垮整组。

掌握 async def 和 await 后，还需要学习如何管理多个协程任务。asyncio 提供了 Task、gather、wait、as_completed、超时和取消等工具。

create_task

asyncio.create_task() 会把协程包装成 Task，并安排它尽快运行。

import asyncio

async def work(name):
    await asyncio.sleep(1)
    return name

async def main():
    task = asyncio.create_task(work('A'))
    result = await task
    print(result)

asyncio.run(main())

如果只创建协程对象而不创建 Task，它不会自动并发执行。

gather

asyncio.gather() 用来并发运行多个 awaitable，并按输入顺序返回结果。

async def main():
    results = await asyncio.gather(
        work('A'),
        work('B'),
        work('C'),
    )
    print(results)

即使 B 最先完成，返回列表仍然按 A、B、C 的顺序排列。

as_completed

如果希望哪个任务先完成就先处理哪个，可以使用 asyncio.as_completed()。

async def main():
    tasks = [work('A'), work('B'), work('C')]
    for task in asyncio.as_completed(tasks):
        result = await task
        print(result)

适合爬虫、批量请求等场景。

wait

asyncio.wait() 可以等待一组任务，并返回完成和未完成的集合。

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

它比 gather() 更底层，适合需要精细控制任务状态的场景。

超时控制

可以使用 asyncio.wait_for() 给任务设置超时。

async def slow():
    await asyncio.sleep(10)

async def main():
    try:
        await asyncio.wait_for(slow(), timeout=1)
    except TimeoutError:
        print('任务超时')

超时后，任务会被取消。

任务取消

async def main():
    task = asyncio.create_task(slow())
    await asyncio.sleep(1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print('任务已取消')

协程内部可以捕获取消异常做清理工作。

async def worker():
    try:
        while True:
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print('清理资源')
        raise

捕获 CancelledError 后通常要重新 raise，否则调用方会以为任务正常完成。

异常处理

gather() 默认遇到异常会抛出。

try:
    await asyncio.gather(task1(), task2())
except Exception as exc:
    print(f'任务失败：{exc}')

如果设置 return_exceptions=True，异常会作为结果返回。

results = await asyncio.gather(task1(), task2(), return_exceptions=True)

并发数量限制

高并发请求时，不应该一次性发出无限任务。可以用 asyncio.Semaphore 限制并发数量。

import asyncio

semaphore = asyncio.Semaphore(3)

async def fetch(url):
    async with semaphore:
        print(f'请求 {url}')
        await asyncio.sleep(1)
        return url

这样同一时间最多只有 3 个协程进入请求区域。

注意事项

create_task() 才会让协程进入调度。
gather() 按输入顺序返回结果。
as_completed() 按完成顺序处理结果。
长时间任务应考虑超时和取消。
大量任务要限制并发数量。

create_task 和直接 await 的区别

直接 await 会等当前协程完成后再继续：

await work('A')
await work('B')

这更像顺序执行。

create_task 会把协程安排到事件循环里，让它可以和其他任务并发推进：

task_a = asyncio.create_task(work('A'))
task_b = asyncio.create_task(work('B'))

await task_a
await task_b

初学时可以记住：想让多个协程“同时开始”，通常要创建任务或使用 gather。

gather 适合一组都要完成的任务

import asyncio


async def fetch(name, delay):
    await asyncio.sleep(delay)
    return f'{name} 完成'


async def main():
    results = await asyncio.gather(
        fetch('A', 2),
        fetch('B', 1),
        fetch('C', 3),
    )
    print(results)


asyncio.run(main())

gather 返回结果的顺序和传入任务的顺序一致，不是完成顺序。

as_completed 适合谁先完成先处理

import asyncio


async def fetch(name, delay):
    await asyncio.sleep(delay)
    return name


async def main():
    tasks = [
        fetch('A', 2),
        fetch('B', 1),
        fetch('C', 3),
    ]

    for done in asyncio.as_completed(tasks):
        result = await done
        print('先拿到：', result)


asyncio.run(main())

如果你在爬取多个页面，谁先回来就先解析谁，as_completed 很合适。

任务异常要认真处理

如果 gather 里的某个任务抛异常，默认会让整个 gather 抛异常：

try:
    await asyncio.gather(task1(), task2())
except Exception as exc:
    print('有任务失败：', exc)

如果你希望收集每个任务的异常，可以用：

results = await asyncio.gather(task1(), task2(), return_exceptions=True)
for result in results:
    if isinstance(result, Exception):
        print('任务失败', result)
    else:
        print('任务成功', result)

不要让后台任务静默失败。看不到异常的并发程序非常难排查。

取消任务不是强制杀死

task.cancel()

取消任务的意思是向任务抛出 CancelledError。任务要运行到下一个 await 附近，才有机会响应取消。

所以协程里可以做必要清理：

async def worker():
    try:
        while True:
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print('收到取消，清理资源')
        raise

捕获后通常要重新 raise，让外部知道任务确实被取消了。

总结

asyncio 任务管理的重点是：启动任务、等待结果、处理异常、控制超时、限制并发。真实项目中，任务管理比单纯写 async/await 更重要。

要点回收

直接 `await` 更像顺序等待；`create_task` 和 `gather` 才能让多个协程并发推进。
`gather` 按传入顺序返回结果，`as_completed` 适合谁先完成先处理。
任务异常和取消都要显式处理，否则异步程序会很难排查。

版权所有

版权归属：Shuo Liu

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