Python基础入门 Day120 异步任务调度：asyncio.create_task 的使用与最佳实践

22次阅读

没有评论

共计 1956 个字符，预计需要花费 5 分钟才能阅读完成。

在前面的内容中，我们学习了协程与异步 I/O。本篇将深入 异步任务调度 ，重点讲解 asyncio.create_task() 的使用方法与最佳实践。

一、什么是 asyncio.create_task？

asyncio.create_task() 用于 显式创建异步任务 ，并将协程添加到事件循环中立即调度执行。

简单来说，它的作用是：

将协程变成“任务”（Task）
让任务立即并发执行，不必等待它完成
实现真正的异步并发结构

二、基本示例：创建独立任务

import asyncio

async def say(name):
    print(f" 开始：{name}")
    await asyncio.sleep(2)
    print(f" 结束：{name}")

async def main():
    task = asyncio.create_task(say(" 任务 1 "))
    print("main 继续运行...")
    await task

asyncio.run(main())

输出说明

main 继续运行... 会在 say() 内部休眠时立即执行
说明：任务已经被并发调度执行。

三、并发创建多个任务

这是 create_task 常见的应用场景：

async def worker(i):
    print(f" 任务 {i} 开始 ")
    await asyncio.sleep(1)
    print(f" 任务 {i} 结束 ")

async def main():
    tasks = []
    for i in range(5):
        task = asyncio.create_task(worker(i))
        tasks.append(task)

    await asyncio.gather(*tasks)

asyncio.run(main())

效果：

五个任务并发执行
总耗时约为 1 秒，而非 5 秒

这是 create_task 的核心优势。

四、create_task 与 gather 的区别

功能	create_task	gather
将协程封装成任务	✔	✖
并发执行多个协程	需要结合其他语法	✔
自动收集结果	✖	✔
控制任务是否独立运行	✔	✖

理解重点：

create_task 是“启动”
gather 是“汇总”

五、独立后台任务（fire-and-forget）

有时我们希望任务在后台运行，不阻塞主逻辑，例如：

async def background_job():
    await asyncio.sleep(3)
    print(" 后台任务完成 ")

async def main():
    asyncio.create_task(background_job())  # 不等待
    print(" 主任务结束 ")

asyncio.run(main())

输出：

主任务会先结束
3 秒后后台任务才输出结果

适用于日志处理、状态更新等无需等待的任务。

六、任务取消：task.cancel()

你可以动态取消一个任务：

async def job():
    try:
        await asyncio.sleep(5)
        print(" 任务完成 ")
    except asyncio.CancelledError:
        print(" 任务被取消 ")

async def main():
    task = asyncio.create_task(job())
    await asyncio.sleep(1)
    task.cancel()
    await task

asyncio.run(main())

输出：

1 秒后 task 被取消
job 内捕获到 CancelledError

适用于超时处理、无效任务清理等场景。

七、任务超时：asyncio.wait_for()

可以设置任务最大等待时间：

async def slow():
    await asyncio.sleep(5)
    return "OK"

async def main():
    try:
        result = await asyncio.wait_for(slow(), timeout=2)
    except asyncio.TimeoutError:
        print(" 任务超时！")

asyncio.run(main())

八、实战：批量启动后台日志任务

import asyncio
import time

async def write_log(i):
    await asyncio.sleep(1)
    print(f" 日志任务 {i} 写入完成 ")

async def main():
    for i in range(5):
        asyncio.create_task(write_log(i))  # 全部后台运行

    print(" 主程序继续运行...")

asyncio.run(main())