Python3.6 Asyncio:类初始化中调用异步函数的最佳实践
Python 3.6 asyncio 类异步初始化的最佳实践
嘿,刚上手asyncio确实容易在类初始化这块犯难,我来给你拆解下可行的方案和最佳实践~
首先明确一个关键点:Python的__init__函数设计上就是同步的,不能直接定义成异步函数——你没法在创建实例的时候用await C(),而且强行把__init__改成async也不符合Python的对象初始化规范,会引发各种奇怪的问题。所以把__init__改成异步函数这条路走不通,咱们换两个靠谱的思路:
方案1:在同步__init__中临时运行事件循环(简单但有局限)
如果你的代码是简单脚本场景,没有预先运行的事件循环,可以在__init__里同步调用事件循环来获取异步函数的返回值。Python 3.6里用asyncio.get_event_loop().run_until_complete()来执行异步任务。
比如结合你提到的aioredis示例:
import asyncio import aioredis from datetime import datetime class C: def __init__(self): self.a = datetime.now() # 同步运行事件循环获取异步结果 loop = asyncio.get_event_loop() self.redis = loop.run_until_complete(self._init_redis()) async def _init_redis(self): # 异步初始化redis连接 return await aioredis.create_redis('redis://localhost') # 使用方式 if __name__ == '__main__': loop = asyncio.get_event_loop() obj = C() # 后续可以在事件循环里用obj.redis做异步操作 loop.run_forever()
⚠️ 注意:这个方案的局限性很大——如果你的代码已经在一个运行中的事件循环里(比如FastAPI、aiohttp这类异步Web框架),再调用run_until_complete会直接报错,因为同一个线程里不能同时运行两个事件循环。所以这种方式只适合简单的独立脚本。
方案2:异步工厂方法(推荐的最佳实践)
这是更符合asyncio异步范式的方案:定义一个异步的类方法(工厂方法),在这个方法里完成所有异步初始化逻辑,最后返回类的实例。
同样用aioredis的例子:
import asyncio import aioredis from datetime import datetime class C: def __init__(self, redis_conn): # 同步初始化已知的属性 self.a = datetime.now() self.redis = redis_conn @classmethod async def create(cls): # 异步逻辑都放在这里 redis_conn = await aioredis.create_redis('redis://localhost') # 创建并返回实例 return cls(redis_conn) # 使用方式 async def main(): # 通过异步工厂方法创建实例 obj = await C.create() # 在这里用obj做异步操作,比如 await obj.redis.set('key', 'value') val = await obj.redis.get('key') print(val) if __name__ == '__main__': loop = asyncio.get_event_loop() loop.run_until_complete(main())
这个方案的优势很明显:
- 完全符合异步编程的逻辑,不会和已有的事件循环冲突
- 把同步初始化和异步初始化逻辑清晰分离,代码可读性更高
- 后续如果需要扩展更多异步初始化步骤,直接在
create方法里加就行
总结
如果是简单的独立脚本,方案1可以快速解决问题;但如果是在异步框架或者复杂的异步环境中,异步工厂方法绝对是最佳选择,它能避免很多潜在的坑,也更符合Python异步编程的设计思想。
内容的提问来源于stack exchange,提问作者qichao_he
相关产品推荐
相关产品推荐

