Python contextvars：协程时代全局变量的正确打开方式

讲个我踩过的真实问题原来写异步日志时候我在模块里放了一个全局request_id单线程跑得好好的一上asyncio.gather就乱了——A 请求还没打完B 请求已经覆盖了变量最后日志里全是乱串的 id。那时候我才意识到协程时代的“全局变量”根本不能用传统思路真正该用的就是contextvars。contextvars是 Python 3.7 起进入标准库的模块专门解决“异步任务之间的状态隔离和传递”。它表面上只是一个变量容器但背后牵扯到Context、Token、协程调度、线程与任务之间的关系甚至还有一套底层 C 实现。这篇文章我不会只讲 API我会把它的原理、asyncio 协作方式、容易踩的坑以及怎么迁移自己的代码一次性拆开讲透。适合正在用 FastAPI、Sanic、Django异步生态或者自己手写异步任务的同学。1. 为什么需要 contextvars1.1 全局变量在并发场景下的失控先看一个最常见的反面例子。request_id None async def business(): info do_query() log.info(business, request_id%s, request_id) return info async def handler_a(): global request_id request_id A-001 await asyncio.sleep(0.1) await business() async def handler_b(): global request_id request_id B-002 await asyncio.sleep(0.1) await business()如果两个 handler 同时跑request_id会被后覆盖的一方污染。A 协程 sleep 完之后business()里读到的可能是 B 的值。用线程模型做隔离的threading.local也一样救不了场同一个线程里可以交替跑多个协程线程不变但协程已经切换了threading.local根本不知道你切换到了哪个任务。问题本质是在 asyncio 里并发的基本单位不是线程而是 Task协程任务。请求进来了、休眠了、恢复了中间可以穿插无数个别的任务。状态隔离必须跟着任务走而不是跟着线程走。1.2 协程世界需要的“上下文”概念contextvars给出的方案是引入一个独立于线程的“上下文”容器。每个ContextVar变量在某个上下文里有一个值不同上下文互不干扰。你可以把它想成每个任务发了一张独立的记录卡任务执行时往自己的卡上写东西切换任务时运行时自动换一张卡。这个“自动换卡”的动作正是 asyncio 调度器帮我们做的。所以用contextvars之后你在协程里设置的值天然只属于当前任务新建子任务时它会复制一份当前卡片作为初始值。这套设计不仅解决隔离问题也解决了父子任务之间“只继承、不乱改”的需求。2. contextvars 核心机制拆解2.1 三个核心对象ContextVar、Context、Tokencontextvars模块里最常见的就是三个东西ContextVar一个变量键。它本身不直接存数据只负责作为字典里的 key。定义时必须给个名字通常还会给一个default。Context保存变量实际值的一组映射。你可以理解成一个 dictkey 是ContextVar对象value 是当前值。Tokenset()操作返回的凭证用来精确撤销这次修改。三者的关系用一句话说ContextVar是钥匙Context是保险柜Token是取回旧状态的收据。Context的实现很有意思。CPython 内部使用了一种叫 HAMTHash Array Mapped Trie的持久化数据结构而不是普通的 dict。这意味着复制上下文时不会把整个 dict 再深拷贝一遍而是共享大部分结构只在需要改动的分支上做复制所以copy_context()的成本很低。这也是它能被 asyncio 高频调用的底气。2.2 set 和 get 背后发生了什么当你执行var.set(value)大概做了三件事拿到“当前上下文”。在当前上下文中写入var - value。返回一个TokenToken 里记录了旧值和这次修改的“锚点”。当你执行var.get()也做三件事拿到“当前上下文”。在这个上下文中找var对应的值。找不到就看ContextVar有没有default有就返回 default没有就抛LookupError。注意get()的查找是只查当前上下文不会像作用域那样一层层往上找。你以为“外层任务设置的值内层任务能读到”是因为子任务在创建时复制了父任务的上下文把初始值带进来了。一旦中途被某个任务 set 覆盖它只影响当前上下文不会影响父任务。import contextvars var contextvars.ContextVar(var, defaultN/A) var.set(hello) print(var.get()) # hello token var.set(world) print(var.get()) # world var.reset(token) print(var.get()) # hello一个很容易忽略的点default不是 set 进去的值它只是 get 兜底用的。如果你在一个上下文里 set 了值然后在子上下文里 get子上下文复制到了这个值所以能读到。如果子上下文从没复制过就会用 default。2.3 Token 到底有什么用reset 为什么不能乱调Token存在的意义是让回滚变得可靠。没有 Token 的话你可能会写成old var.get() var.set(new) # 之后想恢复 var.set(old)这写法在简单场景能跑但只要中间发生了嵌套修改恢复顺序一旦乱了就全乱套。Token 是 set 操作的精确定位reset(token)会把这个变量恢复到 set 之前的状态。但 Token 有个大坑它跟生成它的那个上下文绑定。你在 context A 里 set 得到 token然后跑到 context B 里 reset这种行为是未定义的轻则报错重则污染 B 的数据。所以请记住一句话在同一个上下文里 set 和 reset 必须配对最好用 try/finally 包起来。token var.set(value) try: await do_something() finally: var.reset(token)我用contextvars这么久因为没配对 reset 导致线上请求 id 串掉的 bug 见过不止一次尤其协程里抛异常时忘记 finally 就是重灾区。3. 实操在自己的代码里用好 contextvars3.1 用请求 ID 做一个完整示例最经典的应用就是请求级别的 trace_id。这里我给你一套可以直接抄的写法包括装饰器、日志过滤和异常恢复。import contextvars import uuid import logging request_id_var contextvars.ContextVar(request_id, defaultN/A) def trace_request(func): async def wrapper(*args, **kwargs): token request_id_var.set(uuid.uuid4().hex[:8]) try: return await func(*args, **kwargs) finally: request_id_var.reset(token) return wrapper class RequestIdFilter(logging.Filter): def filter(self, record): record.request_id request_id_var.get() return True logger logging.getLogger(__name__) logger.addFilter(RequestIdFilter())然后在业务代码里你只需要正常调logger.info()日志里就能自动带出request_id。这个方案比手动传参优雅得多尤其当你有一连串内部调用时不用每个函数都塞一个req_id参数。我用这套方案接入过三个项目线上排查问题的效率是肉眼可见的提升。日志里直接搜一个 trace_id整条调用链就串起来了。3.2 父子任务之间的上下文传播规则很多人分不清 asyncio 中子任务和父任务的变量关系。其实规则很简单父协程创建一个 Task 时Task 会复制当前上下文。子 Task 能读到父协程已经设置的值。子 Task 内部 set 的值不会反向影响父协程。两个并行的子 Task 各自独立互相看不见对方的修改。import asyncio import contextvars ctx_var contextvars.ContextVar(ctx_var, defaultbase) async def child(name): print(f{name} before: {ctx_var.get()}) ctx_var.set(name) await asyncio.sleep(0.1) print(f{name} after: {ctx_var.get()}) async def main(): ctx_var.set(parent_value) t1 asyncio.create_task(child(child_A)) t2 asyncio.create_task(child(child_B)) await t1 await t2 print(fparent: {ctx_var.get()}) asyncio.run(main())输出会是child_A before: parent_value child_B before: parent_value child_A after: child_A child_B after: child_B parent: parent_value你看子任务开始读到了父值各自 set 之后不互相干扰父任务也一直保持自己的值。这正是“快照 隔离”的效果。3.3 copy_context() 的正确使用姿势copy_context()会复制一份当前上下文快照返回一个Context对象。你可以用ctx.run(callable)在指定的上下文里执行函数。典型场景是在线程池里执行任务但希望带出主协程的上下文数据。import contextvars import threading var contextvars.ContextVar(var, defaultN/A) var.set(main) def thread_worker(): # 单独线程默认没有 var 的值 print(in thread:, var.get()) t threading.Thread(targetthread_worker) t.start() t.join() # 输出: in thread: N/A # 如果先复制上下文再 run就能读到 ctx contextvars.copy_context() ctx.run(thread_worker) # 输出: in thread: main再强调一次ctx.run()要在合适的线程和上下文中使用不要试图把一个Context对象到处传然后跨线程乱 run。更安全的方式是在目标线程函数开头执行ctx contextvars.copy_context(); ctx.run(worker)让这个线程自己持有快照。这里体现的是copy_context()最大的优点复制成本极低你可以放心频繁调用。4. contextvars 与 asyncio 的协作细节4.1 事件循环怎么做到自动换上下文很多同学以为用了contextvars还需要自己手动把每个 Task 包一层其实完全不需要。asyncio 在 Python 3.7 之后就在 Task 内部内置了上下文管理。流程大致是这样的asyncio.create_task(coro)被调用时Task 内部执行self._context contextvars.copy_context()。当任务开始执行或从休眠恢复时asyncio 内部用self._context.run(...)来包装下一步协程。这样每次协程 step 都运行在任务自己的上下文里。你可以通过task.get_context()Python 3.11取出这个 Task 的 Context 对象自己打印看看。task asyncio.create_task(some_worker()) print(task.get_context())所以“自动传播”不是巧合而是 Task 构造时做的快照 运行时切换上下文配合出来的结果。4.2 快照时机决定了你能继承到哪些值快照发生的时机是create_task()调用时而不是协程定义时。理解这一点对排查问题很有用。var contextvars.ContextVar(var, defaultdefault) async def worker(): print(worker:, var.get()) # 情况1先 set 再 create_task var.set(aaa) t1 asyncio.create_task(worker()) # worker 读到 aaa # 情况2先 create_task 再 set t2 asyncio.create_task(worker()) var.set(bbb) await t2 # worker 可能读到 default 或旧值取决于调度顺序通常不会是 bbb也就是说如果你想给一批子任务统一初始化某个值一定要在创建任务之前 set 完毕而不是创建之后再 set。4.3 踩坑不要把 Token 跨任务传递这是最容易忽略的隐蔽问题。Token是绑定到具体上下文的它只在生成它的那个上下文里有效。如果你在父任务里 set 得到 token然后把它传给子任务去 reset常见结果是 ValueError 或者离谱的变量污染。子任务如果想临时修改变量就在子任务自己的上下文里 set然后立刻记录自己的 token再 finally reset。另外一个常见的坑是asyncio.wait_for。它内部会对协程再包一层任务导致上下文快照可能比你预期的层次更深。别指望在wait_for的外层 set 某个值内层一定能按你的直觉传播最好的做法仍然是在顶层业务逻辑开始时设置好上下文值后续不要依赖中间层再单点注入。4.4 跨线程时千万别乱来contextvars虽然看起来像是一张“全局卡片”但它默认是绑定到当前线程的内存结构里的。把主线程的Context对象直接扔到另一个线程去run()是不安全的。我在项目里遇到过崩溃原因就是在线程池中误用了共享的ctx对象。正确做法是让每个线程在入口时自己复制上下文。def thread_entry(): current_ctx contextvars.copy_context() current_ctx.run(real_work)这样线程内部有独立的快照既不会污染主线程也不会因为共享同一个可变对象而出现难以复现的数据竞争。5. 常见易错点与排查技巧5.1 一张表看懂常见问题场景现象原因解决方法模块级全局变量存请求信息并发日志串号普通变量没有隔离能力换成 ContextVar新线程读取主线程 ContextVar拿到默认值线程之间默认不继承上下文线程入口 copy_context() 再 run忘写 default 参数抛 LookupErrorget() 找不到值定义时给 default 或用 get(default)协程内 set 没 reset任务复用时污染Token 被覆盖或丢失try/finally 配对 reset子任务修改值影响父任务认为上下文是全局的子任务快照是独立副本理解传播规则在中间层临时 set 上下文后续代码读到不期望值修改作用域超出预期最小化 set 范围用装饰器/上下文管理器这些踩坑点我基本都遇到过尤其是第一条属于异步架构改造时期的高频问题。如果你正好在做 FastAPI 或爬虫框架改造建议先自查一遍。5.2 如何定位上下文丢失问题当你怀疑某个ContextVar没传到目标协程第一件事是打印当前上下文的快照。import contextvars ctx contextvars.copy_context() for item in ctx.items(): print(repr(item[0].name), , repr(item[1]))Context对象可以迭代key 是ContextVarvalue 是当前值。看到 items 列表之后你就能立刻判断是压根没 set还是 set 在了别的上下文里还是被覆盖了。如果想确认两个任务是不是同一个上下文也可以比较id(task.get_context())。异步任务之间正常应该不同如果相同说明你的代码在某处绕过了 asyncio 的任务包装逻辑用了裸协程或手动 context 切换。5.3 调试时的建议给日志专门增加一个request_id字段比你在代码里到处打印变量值要直观得多。我常用的做法是三层入口中间件设置trace_id到contextvars。logging 的 Filter 读取trace_id注入日志记录。全局异常处理里也读trace_id方便把异常和请求关联起来。调试时不用断点直接看日志字段所有跨任务的上下文传播问题基本能一目了然。另外别在一个业务函数里反复 set 同一个变量每 set 一次就会产生一个新 token管理成本直线上升。能设置一次的地方就不要设置第二次。6. 与 ThreadLocal、显式传参等方案对比6.1 不同方案的差异方案隔离单位能否跨线程能否跨协程代码侵入度性能开销普通全局变量无共享共享低极低ThreadLocal线程隔离不隔离中低显式传参函数调用链可传可传高低contextvars上下文/Task需手动复制自动传播低略高于直接变量可以看到contextvars在 asyncio 生态里是唯一一个既做到自动隔离又不需要把参数疯狂传下去的方案。代价是它引入了“隐形状态”调用链上不直观因此需要开发者在代码设计上保持克制。6.2 什么时候别用 contextvars不是所有地方都适合用。如果你的业务函数之间参数本来就清晰数据量又小那直接传参可能是最可靠的方案。contextvars适合那些“横切关注点”比如 trace_id、user_id、权限上下文、数据库事务标识。如果每个业务函数都要读取这些信息与其层层传参不如放上下文里。但也不建议往里塞大量业务数据。我在项目里看到有人把一个大对象塞进 ContextVar比如把整个数据库 session 放进去虽然方便但是一旦忘记 reset内存引用生命周期会被意外拉长。请把它当成一个轻量级的请求作用域容器而不是万能数据仓库。6.3 我最终选型时的判断标准我做异步项目时的判断顺序如下这个数据是不是每个处理环节都要用如果是优先 contextvars。是不是只需要在某一条调用链里传递如果是优先显式传参。是不是线程和协程都会有如果是统一在协程入口处 copy_context()然后让线程内部分发。是不是只需要在当前任务内临时用如果是用 set/token/finally 包住局部作用域。按这个标准来contextvars的引入不会导致代码失控。其实很多团队不敢用它是怕“隐式依赖”但这种担心可以通过规范代码结构解决只在入口和出口用中间业务代码一律只读不写。我自己的项目现在把contextvars用得很顺手。日志里带上了 trace_id用户登录态通过认证中间件注入上下文后续业务函数不需要再传 user_id。排障时看一条日志就能还原整个请求链路效果比之前用全局变量时好了不止一个量级。踩过几次跨线程和 token 的坑之后我现在的经验就是set 之后一定要无条件 reset线程之间不要共享同一个 Context 对象凡是进入子任务的地方都只读变量、不随意覆盖。能做到这三条contextvars就是你进行异步开发时的稳定后盾。