asyncio环境下传统单例失效,需用模块级变量+asyncio.Lock实现懒初始化单例;锁必须全局唯一,contextvars不适用全局场景,多loop时单例边界为每个event loop内唯一。

asyncio 环境下不能直接用类装饰器或模块级实例
传统单例(比如 @singleton 装饰器或 __new__ 拦截)在 asyncio 中会失效——因为多个协程可能并发进入初始化逻辑,导致竞态,最终创建多个实例。更隐蔽的问题是:同一个 asyncio.EventLoop 里,不同协程调用同一类的构造函数时,__new__ 不会自动同步,除非你手动加锁。
实操建议:
- 避免在
__new__或__init__里做异步初始化(比如 await DB 连接),那会导致语法错误 - 把“单例”和“异步初始化”拆开:先确保对象实例唯一,再用 lazy async 初始化方法补充状态
- 最稳妥的方式是用模块级变量 +
asyncio.Lock控制首次初始化
推荐方案:模块级实例 + 异步 init_once 方法
核心思路是把单例对象放在模块顶层,用一个带锁的 init_once 协程保证只初始化一次,后续调用直接返回已初始化对象。这不是“懒汉式单例”,而是“懒初始化单例”。
示例结构:
立即学习“Python免费学习笔记(深入)”;
import asyncio
<p>_lock = asyncio.Lock()
_instance = None</p><p>class AsyncSingleton:
def <strong>init</strong>(self):
self.db_conn = None # 占位,不 await</p><pre class="brush:php;toolbar:false;">async def init(self):
if self.db_conn is not None:
return
# 模拟耗时异步初始化
await asyncio.sleep(0.1)
self.db_conn = "fake-connection"async def get_singleton(): global _instance if _instance is None: async with _lock: if _instance is None: # double-check _instance = AsyncSingleton() await _instance.init() return _instance
注意:_lock 必须是模块级 asyncio.Lock(),不能每次调用都新建;否则锁失效。
为什么不用 aiocontextvars 或 task-local?
contextvars.ContextVar 是任务局部的,不是全局单例——每个 asyncio.Task 都能 set 自己的值,无法跨协程共享同一对象。如果你要的是“每 Task 一个实例”,那 contextvar 合适;但你要的是“整个 event loop 里仅一个实例”,它就不适用。
常见误用现象:
- 用
ContextVar().set(obj)后,在另一个协程里get()返回None - 误以为
asyncio.current_task()能当单例容器,其实它只标识当前任务 - 试图用
asyncio.get_event_loop()当 key 存对象,但 Python 3.12+ 已弃用该方法,且多线程 loop 场景下不可靠
进阶注意:多线程 + 多 loop 场景怎么办?
如果程序用 asyncio.run() 多次启动(如测试中反复调用),或用了 asyncio.new_event_loop(),那么每个 loop 都会拥有独立的模块作用域,_instance 不共享。此时单例边界其实是“每个 event loop 内唯一”,而非进程级唯一。
若真需进程级单例(比如共享一个 Redis 连接池),必须引入外部协调机制:
- 用
multiprocessing.Manager或threading.local+ 全局锁(但会破坏纯异步语义) - 改用连接池类(如
aioredis.Redis本身已内置连接复用),不再自己造单例 - 把单例逻辑下沉到 C 扩展或共享内存(极少数场景)
绝大多数业务场景,“每个 event loop 一个单例”已足够,强行追求进程级唯一反而增加复杂度和死锁风险。


















