contextvars.ContextVar 解决 asyncio 中协程间变量隔离问题,因 threading.local 在单线程协程切换时失效,global 变量被共享导致竞态;它提供每个 Task 独立、可继承、自动随 await 流转的上下文变量空间。

contextvars 在 asyncio 任务中为什么不能直接用 global 或 threading.local?
因为 asyncio 任务在同一个线程内协程切换,threading.local 完全失效;而 global 变量会被所有任务共享、相互覆盖。比如你用 global request_id,A 任务刚设完值,await 切到 B 任务改了它,A 恢复后读到的就是 B 的值——这根本不是“上下文”,是竞态。
真正需要的是:每个 Task 拥有自己独立、可继承、自动随 await 流转的变量空间。这就是 contextvars.ContextVar 的设计目标。
如何正确创建和设置 ContextVar 并让它跨 await 传递?
关键在于:必须在任务启动前(或 Task 创建时)显式拷贝当前上下文,否则子任务会拿到空上下文。常见错误是直接在 async def 里调 var.set(),但没绑定到当前 task 上下文。
-
var = contextvars.ContextVar('request_id', default=None)—— 声明时务必设default,否则var.get()会抛LookupError - 在任务入口(如
asyncio.create_task()调用点)用asyncio.Task.current_task().get_coro().__context__不可靠;正确做法是用contextvars.copy_context()+loop.create_task(coro, context=ctx)(Python 3.11+) - 更兼容的做法:在父协程里
var.set(value),然后直接await sub_coroutine()—— 因为await本身会继承当前上下文;但create_task()不会自动继承,必须手动传
create_task 时怎么把父上下文带进去?(Python < 3.11)
低版本 Python 没有 context 参数,得靠 asyncio.run_coroutine_threadsafe 或封装一层。最稳妥的实操方式是:用 contextvars.copy_context() 拿到当前上下文,再通过 functools.partial 把变量值注入子协程参数,或者用闭包捕获。
立即学习“Python免费学习笔记(深入)”;
图片提示词生成器?不止如此。 马甲系统 —— 把脑海中的画面,翻译成AI能理解的专业表达。 用得越多,它越懂你:首次需要多问几句确认方向,用久了几乎一说就懂。 用得越多,它越快:缓存机制让后续对话越来越省。 RAG进化:成功案例持续入库,越跑越聪明。 输入「新手指南」查看完整功能介绍
但更轻量的方案是:在 task 启动前,用 context.run() 显式运行子协程:
ctx = contextvars.copy_context() task = asyncio.create_task(ctx.run(sub_coro))
注意:ctx.run() 返回的是协程对象,不是结果,所以要包一层 asyncio.create_task();且 sub_coro 必须是普通协程函数,不能是已 await 过的 coroutine object。
为什么 logging 或中间件里 get() 总是返回 default?
大概率是上下文断掉了:比如用了 loop.run_in_executor(进线程池)、调了 concurrent.futures、或在 signal handler / atexit 里访问 —— 这些场景都脱离了 asyncio event loop 的上下文管理。还有种情况是:你在非 async 函数里(比如同步装饰器)调了 var.get(),此时根本没有 ContextVar 绑定。
- 检查是否在
ThreadPoolExecutor中 —— 需手动把contextvars.copy_context()传进去,再在子线程里ctx.run() - 避免在
__del__、atexit、signal.signal回调中读取ContextVar - 用
asyncio.current_task()是否为None快速判断当前是否在 task 上下文中
上下文不是魔法,它只在 async/await 协程链路里自动流转;一旦跨出这个边界,就得靠你自己搬运。

















