Python 3.11.3中asyncio.create_task()无法后台运行函数问题排查

• loop.run_forever()会持续阻塞当前线程，直到主动调用loop.stop()，所以它后面的print("done")永远不会执行，这和你预期的输出直接冲突。
• 你用asyncio.create_task()创建的infinite_async_func确实已经在后台运行了，但事件循环被run_forever无限占用，根本没机会执行后续代码。

方案一：让事件循环运行一段时间后退出，输出done #

用Python 3.7+推荐的asyncio.run()代替手动管理事件循环，同时让主协程等待一段时间后主动退出，这样后台任务会在这段时间内运行，之后事件循环结束就能执行print("done")：

import asyncio
import nest_asyncio

nest_asyncio.apply()

async def infinite_async_func():
    while True:
        await asyncio.sleep(1)
        print("infinite_async_func is running")

async def main():
    print("Main is running")
    # 创建后台任务
    task = asyncio.create_task(infinite_async_func())
    # 让主协程运行3秒，模拟业务逻辑
    await asyncio.sleep(3)
    # 取消后台任务，避免事件循环关闭时报错
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("后台任务已取消")

if __name__ == "__main__":
    asyncio.run(main())
    print("done")

输出示例：

Main is running
infinite_async_func is running
infinite_async_func is running
后台任务已取消
done

方案二：保持后台任务长期运行，同时主程序执行其他逻辑 #

如果需要后台任务一直运行，同时主程序能处理其他同步逻辑，可以用线程分离事件循环：

import asyncio
import nest_asyncio
import threading

nest_asyncio.apply()

async def infinite_async_func():
    while True:
        await asyncio.sleep(1)
        print("infinite_async_func is running")

async def background_task():
    await infinite_async_func()

def run_event_loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(background_task())

if __name__ == "__main__":
    print("Main is running")
    # 启动守护线程运行事件循环和后台任务
    thread = threading.Thread(target=run_event_loop, daemon=True)
    thread.start()
    # 主程序立即输出done，还可以加其他同步逻辑
    print("done")
    # 让主程序保持运行，否则后台线程会随主进程终止
    input("按回车键退出...\n")

输出示例完全符合你的预期：

Main is running
done
infinite_async_func is running
infinite_async_func is running
...

• async with是异步上下文管理器语法，和你当前的后台任务需求无关，不需要用到。
• 如果你的代码是在普通Python环境运行，nest_asyncio可以直接去掉，它主要是给Jupyter这类已有事件循环的环境用的。

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