Python中如何并行初始化类对象并等待所有__init__方法执行完成

Python 中类的__init__是同步原生方法，无法直接被await，也不能在内部执行异步逻辑实现并发。我们可以将耗时的初始化逻辑拆分为异步方法，配合 asyncio 实现并发初始化，总耗时仅为单次初始化的60秒，完全符合预期输出要求。

完整可运行代码 #

import asyncio

class First:
    def __init__(self, x):
        self.x = x
        print("First Class")
    
    async def initialize(self):
        await asyncio.sleep(60)
        print("First Class done")

class Second:
    def __init__(self, x):
        self.x = x
        print("Second Class")
    
    async def initialize(self):
        await asyncio.sleep(60)
        print("Second Class done")

class Third:
    def __init__(self, x):
        self.x = x
        print("Third Class")
    
    async def initialize(self):
        await asyncio.sleep(60)
        print("Third Class done")

async def main():
    # 先同步创建所有实例，立刻输出三个Class日志
    f = First(1)
    s = Second(2)
    t = Third(3)
    # 并发执行所有异步初始化逻辑，等待全部完成
    await asyncio.gather(
        f.initialize(),
        s.initialize(),
        t.initialize()
    )

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

代码说明 #

• 替换原有的同步time.sleep为asyncio.sleep，避免阻塞事件循环导致无法并发
• 轻量的同步属性赋值逻辑保留在__init__中，耗时逻辑抽为独立的异步initialize方法
• 先同步创建三个类的实例，此时会立刻打印三个类的初始化启动日志
• 通过asyncio.gather并发执行所有实例的异步初始化方法，等待全部执行完成后再退出主函数

可选方案：异步工厂方法 #

如果希望实例返回时就已经完成所有初始化、避免拿到未就绪的实例，可以用异步类方法实现：

class First:
    def __init__(self, x):
        self.x = x
    
    @classmethod
    async def create(cls, x):
        print("First Class")
        await asyncio.sleep(60)
        print("First Class done")
        return cls(x)

# 其余Second、Third类同理修改

async def main():
    # 并发创建所有实例，全部初始化完成后返回
    instances = await asyncio.gather(
        First.create(1),
        Second.create(2),
        Third.create(3)
    )

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