为何调用asyncio.create_task()后异步任务未立即执行？

测试代码 #

import asyncio

async def worker():
    print("worker start")
    await asyncio.sleep(1)
    print("worker done")

async def main():
    # Schedule the coroutine
    asyncio.create_task(worker())
    # Give control back to the event loop for a moment
    await asyncio.sleep(0.1)
    print("main exiting")

asyncio.run(main())

预期行为 #

原本以为调用asyncio.create_task(worker())会立即启动worker()，预期输出：

worker start
main exiting
worker done

实际行为 #

实际输出中，worker()的内容在main()执行完毕后才出现：

main exiting
worker start
worker done

已尝试的解决方法 #

• 延长await asyncio.sleep(0.1)的延迟时间
• 将asyncio.create_task替换为await worker()
• 在Python 3.11和3.12版本中运行（结果一致）

问题 #

1. 为何asyncio.create_task()调用后任务未立即开始执行？
2. 有没有推荐的模式，可以确保任务在main()退出前获得运行机会（无需直接await该任务）？

环境信息 #

• Python 3.12.2
• macOS 14.4（Sonoma）

1. 任务未立即执行的原因 #

asyncio.create_task()只是把协程包装成异步任务，加入事件循环的待执行队列，不会打断当前协程的执行流程。asyncio采用协作式调度机制，只有当当前协程主动让出控制权（比如执行await操作、调用asyncio.sleep(0)强制切换）时，事件循环才会从队列里取出任务调度执行。

在你的代码里，调用create_task(worker())后，main协程会继续往下跑，直到遇到await asyncio.sleep(0.1)。此时事件循环会挂起main协程等待定时器到期，但这段时间里，事件循环并没有优先调度worker任务（可能是定时器回调的优先级高于普通任务）。等0.1秒到了，main协程恢复，先打印main exiting再结束。此时asyncio.run在关闭事件循环前，会处理剩余的待执行任务，worker才开始运行，所以你看到main exiting先输出。

2. 确保任务在main退出前运行的推荐模式 #

方式一：强制让出控制权

在创建任务后，立即await asyncio.sleep(0)，这会触发事件循环切换到worker任务执行：

async def main():
    asyncio.create_task(worker())
    await asyncio.sleep(0)  # 强制让事件循环调度worker
    await asyncio.sleep(0.1)
    print("main exiting")

方式二：使用TaskGroup（Python 3.11+）

这是最推荐的模式，TaskGroup会自动等待所有子任务完成，不需要显式await单个任务，代码更简洁安全：

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(worker())
    # 所有子任务完成后才会执行到这里
    print("main exiting")

方式三：批量等待任务

如果不想用TaskGroup，可以用asyncio.wait()或asyncio.gather()批量管理任务，在main退出前等待所有任务完成：

async def main():
    task = asyncio.create_task(worker())
    await asyncio.sleep(0.1)
    print("main exiting")
    await asyncio.wait([task])  # 等待worker任务完成

内容的提问来源于stack exchange，提问作者YANG Zijie