如何以Pythonic方式实现需异步初始化的单例类？

要实现符合需求的异步单例类，核心是在类级别维护唯一实例，并通过异步锁避免并发创建冲突，同时兼容用户传入ClientSession或自动创建的场景。以下是修改后的完整代码：

import aiohttp
from asyncio import Lock

class AsyncFoo:
    _instance = None
    _lock = Lock()  # 异步锁，防止多协程并发创建实例
    client_session: aiohttp.ClientSession

    def __init__(self):
        raise NotImplementedError("Direct instantiation is not allowed. Use `AsyncFoo.create()`.")

    @classmethod
    async def create(cls, session: aiohttp.ClientSession | None = None):
        # 双重检查锁：先快速判断实例是否存在，减少锁竞争
        if cls._instance is not None:
            return cls._instance
        
        async with cls._lock:
            # 进入锁后再次检查，避免锁等待期间实例已被创建
            if cls._instance is not None:
                return cls._instance
            
            instance = cls.__new__(cls)
            if session is None:
                instance.client_session = aiohttp.ClientSession()
                # 标记这是内部创建的session，方便后续清理
                setattr(instance.client_session, '_internal_session', True)
            else:
                instance.client_session = session
                setattr(instance.client_session, '_internal_session', False)
            cls._instance = instance
            return instance

    async def send_request(self):
        # 注意：异步HTTP请求必须用await调用
        await self.client_session.post("https://example.com/api")
    
    async def close(self):
        # 仅关闭类内部创建的session，避免关闭用户传入的外部session
        if getattr(self.client_session, '_internal_session', False):
            await self.client_session.close()
        # 重置实例，允许后续重新创建
        AsyncFoo._instance = None

    @classmethod
    async def get_instance(cls):
        # 提供无参数的快捷方法，方便用户直接获取实例
        if cls._instance is None:
            return await cls.create()
        return cls._instance

关键实现细节 #

• 单例核心机制：通过类变量_instance存储唯一实例，所有create或get_instance调用都会返回同一个对象。
• 异步安全：使用asyncio.Lock确保在多协程环境下，只有一个协程能进入实例创建逻辑，避免重复初始化。
• Session兼容性：
  • 第一次调用create时传入的session会被永久绑定到单例实例，后续调用create时传入的session参数会被忽略。
  • 自动创建的session会被标记，在调用close()时自动清理，不会影响用户传入的外部session。
• 资源清理：提供close()方法，既可以清理内部创建的session，又能重置单例状态，允许后续重新创建实例。

使用示例 #

import asyncio

async def main():
    # 场景1：传入自定义session
    custom_session = aiohttp.ClientSession()
    foo1 = await AsyncFoo.create(custom_session)
    
    # 场景2：自动创建session
    foo2 = await AsyncFoo.get_instance()
    assert foo1 is foo2  # 验证单例：返回同一个实例
    
    # 发送请求
    await foo1.send_request()
    
    # 清理资源（可选）
    await foo1.close()

asyncio.run(main())

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