应使用 contextvars.contextvar 替代 threading.local,其通过 .set()/.get() 实现协程级隔离,需在模块顶层定义、手动 reset 且不跨线程自动传递。

contextvars 在异步任务里丢变量怎么办
用 asyncio 时,threading.local 完全失效——因为协程在同一线程内切换,local 存的值会被覆盖或错乱。这时候必须换 contextvars。
典型场景:HTTP 请求来了,你想把 request_id 注入到整个调用链(日志、DB 查询、下游调用),又不想一层层手动传参。
-
ContextVar实例必须在模块顶层定义,不能在函数里重复创建(否则每次都是新变量) - 读写都得通过
.get()和.set(),直接赋值没用 - 异步函数里,
set()后的值只对当前Task可见,不会污染其他并发任务
import contextvars
req_id_var = contextvars.ContextVar('request_id', default=None)
<p>async def handle_request():
token = req_id_var.set('req-123') # 绑定到当前上下文
try:
await do_work()
finally:
req_id_var.reset(token) # 必须 reset,尤其在异常路径下为什么 ContextVar.get() 返回 default 而不是报错
这是设计使然:ContextVar 不强制要求“必须已设置”,它更像一个带默认值的可选上下文槽位。很多框架(如 Starlette)靠这个特性做“有则用、无则跳过”的柔性注入。
但这也容易掩盖问题:你以为值已经设好了,结果 .get() 拿到的是 default,日志全打成 None 或空字符串,排查起来很隐蔽。
立即学习“Python免费学习笔记(深入)”;
- 上线前务必检查所有
.get()调用点,是否该加非空断言(比如req_id_var.get() or raise RuntimeError) - 别依赖
default=None来判断“未初始化”,而要用ContextVar的.get(token)配合token是否有效来判断 - 测试时故意不
set(),看下游逻辑是否静默失败
sync 函数里混用 contextvars 的坑
同步代码跑在异步任务里(比如 await loop.run_in_executor(...)),或者用了多线程(ThreadPoolExecutor),ContextVar 值默认不会自动传递过去。
这不是 bug,是明确设计:上下文是协程级的,不是线程级的。跨 executor 时,你得手动搬运。
- 用
contextvars.copy_context()拿到当前上下文快照,再在子线程里用ctx.run(...)执行函数 - 别试图在子线程里直接调
var.get()——大概率拿到default - 第三方库(如
concurrent.futures)不自动支持,除非它显式做了上下文透传(比如anyio.to_thread.run_sync就做了)
和 logging.LoggerAdapter 比,contextvars 有什么不可替代性
LoggerAdapter 只能影响日志输出,而 contextvars 是通用上下文载体:你可以用它驱动重试策略(按 user_tier 变超时)、控制采样率(trace_sample_rate)、甚至决定 DB 连接池(tenant_id)。它不是日志工具,是轻量级请求作用域容器。
但代价是:它不提供自动注入能力。你得自己在入口(ASGI middleware、FastAPI dependency)里 set(),也得自己确保每个出口(异常处理、finally 块)里 reset()。
- 不要把它当全局配置用——
ContextVar生命周期绑定到单次请求/任务,跨请求复用会出错 - 避免在循环里反复
set()+reset(),性能差;高频场景建议用局部变量+参数传递 - 调试时可用
contextvars.copy_context().items()快速 dump 当前上下文,比扒堆栈快得多
最麻烦的其实是 reset 的时机:漏掉一次,那个值就可能漂移到下一个无关任务里去,而且只在高并发压测时才偶然复现。










