如何在类内部创建实例时启动异步任务？

我有如下代码：

import asyncio
class Test:
    async def hello_world(self):
        while True:
            print("Hello World!")
            await asyncio.sleep(1)
test = Test()
loop = asyncio.get_event_loop()
# Blocking call which returns when the hello_world() coroutine is done
loop.run_until_complete(test.hello_world())
loop.close()

请问是否可以在类内部实现创建类实例时自动启动hello_world异步函数，无需从外部调用？

当然可以！有几种实用的方案能满足你的需求，下面给你详细说明：

方案一：在__init__构造方法中启动异步任务 #

你可以在类的构造方法里获取当前事件循环，用asyncio.create_task把hello_world协程包装成异步任务并提交到循环中。这样实例一创建，任务就会自动开始运行。

修改后的代码示例：

import asyncio

class Test:
    def __init__(self):
        # 获取当前活跃的事件循环
        self.loop = asyncio.get_running_loop()
        # 创建并启动异步任务
        self.task = self.loop.create_task(self.hello_world())

    async def hello_world(self):
        while True:
            print("Hello World!")
            await asyncio.sleep(1)

async def main():
    test = Test()  # 实例化瞬间自动启动hello_world
    await asyncio.sleep(5)  # 让程序运行5秒后结束
    test.task.cancel()  # 手动取消任务，避免程序退出时抛出异常

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

这里用asyncio.run()是Python 3.7+推荐的异步程序启动方式，它会自动创建、管理和关闭事件循环。另外记得在合适的时机取消任务，也可以在hello_world里捕获asyncio.CancelledError来做优雅的退出处理。

方案二：使用异步类方法创建实例（更适合复杂异步场景） #

如果你的类初始化本身需要依赖其他异步操作（比如异步连接数据库、读取远程配置），推荐用类方法来封装实例创建和任务启动的逻辑，这样能避免在__init__里处理异步逻辑的潜在问题：

import asyncio

class Test:
    def __init__(self):
        self.task = None

    @classmethod
    async def create(cls):
        self = cls()
        self.task = asyncio.create_task(self.hello_world())
        # 这里还可以添加其他异步初始化逻辑，比如await self.init_db()
        return self

    async def hello_world(self):
        while True:
            print("Hello World!")
            await asyncio.sleep(1)

async def main():
    test = await Test.create()  # 通过异步类方法创建实例并启动任务
    await asyncio.sleep(5)
    test.task.cancel()
    try:
        await test.task
    except asyncio.CancelledError:
        print("异步任务已被优雅取消")

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

这种方式的逻辑更清晰，扩展性也更强，适合实际项目中的复杂异步场景。

小提示 #

• 如果你使用Python 3.6及以下版本，asyncio.get_running_loop()需要替换为asyncio.get_event_loop()，因为前者是Python 3.7新增的API。
• 一定要管理好异步任务的生命周期，避免程序退出时出现未处理的任务取消异常。

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