做后端这些年我最怕的不是某个接口写错了而是“串号”——两个请求的数据互相干扰。去年排查一个生产事故用户A突然看到了用户B的购物车最终定位到问题根源有人在 async 任务里用了 threading.local。那一刻起我决定把 contextvars 彻底弄明白。contextvars 是 Python 3.7 起进入标准库的模块解决的是在协程、Future、Task 交织执行时怎么安全地把 request_id、用户信息、事务 ID 这类上下文传递给后续代码。它比 threading.local 更适合异步场景asyncio、FastAPI 等框架的链路追踪底层也离不开它。这篇文章不讲官方文档翻译只从原理和实战踩坑出发把 Context、ContextVar、Token 这套东西彻底拆开。1. 认清问题为什么 threading.local 在异步环境翻车1.1 一次“串号”事故的完整经过那次事故的背景大概是这样的一个 FastAPI 服务内部大量使用asyncio.create_task做并行 IO。当时为了把当前登录用户传给内部函数代码里直接用了threading.local。压测阶段多个请求同时进来事件循环在单线程里来回切换线程 id 始终是同一个threading.local的存储区也是同一个。结果请求 A 刚把user_id写进去协程挂起请求 B 切换到同一线程把user_id覆盖成自己的请求 A 恢复执行后一读就变成了 B 的用户。这就是典型的变量串号。threading.local的底层实现是按线程 ID 作为 key 隔离数据的它默认“一个线程同一时刻只处理一件事”。这个假设在同步阻塞模型里成立但在事件循环里完全失效——一个线程可以交叉执行成百上千个协程每个协程都代表一个独立的逻辑任务。只靠线程维度去隔离自然就串了。再往下说一点很多人用threading.local是因为它的 API 太友好local.user_id 1、local.user_id读出来就行。可一旦代码里混入 async这种“友好”就变成了陷阱。真正排查的时候你根本没法从堆栈里看出是哪个协程污染了数据因为线程号一模一样日志里全是同一串数字。1.2 contextvars 的设计思路把隔离维度从“线程”换成“上下文”contextvars的核心思想是每个正在运行的逻辑上下文协程、Task、请求处理链都维护一份独立的变量快照。你用 ContextVar 设置的值只会影响当前上下文当 asyncio 切换到另一个任务时会自动换上那个任务自己的上下文。换句话说threading.local按“房间”隔离contextvars 按“房间里的会话”隔离。这个设计对日常使用是透明的。你不需要每次手动copy_context、run这些东西绝大多数场景只要var.set()和var.get()。asyncio 在任务创建和切换时会自动接管上下文的保存和还原。这也是它能成为日志中间件、链路追踪库底层依赖的原因。对比维度threading.localcontextvars隔离粒度线程上下文协程/任务/请求单线程协程场景会串号不会是否区分多个异步任务不能每个任务独立快照框架支持需要自己维护asyncio 原生支持2. 核心结构拆解ContextVar、Context 与 Token2.1 两个“容器”、一个“凭证”ContextVar是一个变量的定义者。它本身不保存任何值只是一把“钥匙”。你在任何地方用同一把钥匙去 get 或 set操作的其实都是“当前上下文”里的对应格子。Context是一组 ContextVar 值的集合。每个异步任务、每个请求链路都可以拥有一个独立的 Context。不同的 Context 里同一个 ContextVar 可以关联不同的值。比如任务 A 的 Context 里user_idu-1001任务 B 的 Context 里user_idu-1002两者互不干扰。Token是 set 之后返回的凭证。它记录“改之前是什么值”reset 时按这个凭证恢复现场。这三个东西合起来就是 contextvars 最小的运行模型。2.2 从 set 到 reset完整生命周期直接看一段代码import contextvars session_id contextvars.ContextVar(session_id, defaultEMPTY) print(session_id.get()) # EMPTY token session_id.set(u-1001) print(session_id.get()) # u-1001 session_id.reset(token) print(session_id.get()) # EMPTYContextVar创建时有两个参数名字和默认值。get()时当前上下文里如果没有这个变量就会返回默认值。注意默认值并不会在创建时写入 Context它是在get()时按需使用的。这个细节很多人第一次会忽略后面排查“为什么拿到默认值而不是我 set 的值”时往往就卡在这里。嵌套 set 的场景更接近真实业务t1 session_id.set(first) t2 session_id.set(second) print(session_id.get()) # second session_id.reset(t2) # 回退到 first print(session_id.get()) # first session_id.reset(t1) # 回退到 EMPTY print(session_id.get()) # EMPTYToken 内部大致保存了三个信息绑定了哪个 ContextVar、旧值是多少、这个 token 有没有被消费过。重复 reset 同一个 token 会抛RuntimeError: Token has already been used once.因为它已经被“使用过一次”了。reset 本质上是把旧值写回而不是简单删除键。2.3 底层数据结构一份哈希表的存取CPython 中_contextvars是 C 实现的模块Python 侧只是薄封装。一个 Context 本质上是一个哈希表key 是 ContextVar 对象的内部标识value 是当前值。因此单次 get 和 set 的时间复杂度都是 O(1)。这一点很关键它意味着高频任务切换时上下文开销不会变成性能瓶颈。这里有个容易忽略的点copy_context()是对这个哈希表的浅拷贝。它拷贝的是“表里有哪些 key 和 value 引用”不会递归拷贝每个 value 对象。所以复制出来的 Context 在读取已有变量时开销很小但如果你在值对象内部做了修改要意识到它和原 Context 指向的是同一个对象。2.4 一个容易误导人的点多次 copy_context每调用一次copy_context()得到的都是一个独立的 Context 快照。它和当前上下文、以及其他快照之间互不干扰。你在快照 A 里 set 某个 var不会影响当前上下文也不会影响快照 B。这个特性是“向子任务传播上下文”行为的基础。举个例子当前 Context 里var 1你ctx contextvars.copy_context()然后在ctx.run(某个函数)里面把 var 改成 2。函数执行结束后外部的当前 Context 里 var 仍然是 1。这听起来简单但很多人会把copy_context()和“引用同一个上下文”搞混误以为改一处会影响所有处。注意ContextVar 的默认值在 get() 时按需使用不会在创建时写入任何 Context。理解这个点排查“为什么 get 到的是默认值而不是我 set 的值”会少走很多弯路。3. 事件循环里的流转asyncio 与 contextvars 的合作3.1 create_task 时发生了快照直接看代码import asyncio import contextvars trace_id contextvars.ContextVar(trace_id, default-) async def child(): print(child sees:, trace_id.get()) await asyncio.sleep(0) async def main(): trace_id.set(t-001) task asyncio.create_task(child()) await task asyncio.run(main()) # child sees: t-001当调用asyncio.create_task(child())时事件循环会先 copy 当前上下文然后把这个快照绑到新 Task 对象上。等到 task 真正被调度执行时循环会把当前上下文切到这个快照child 协程里所有ContextVar.get()就会从快照里取。所以这里有一个非常重要的推论子任务能看到的上下文是“创建这个子任务那一刻”的上下文。所有在create_task之前完成的 set子任务都能看到之后新 set 的子任务一律看不到。很多人误以为只要父任务 set 了子任务任何时候都能读其实不是。3.2 Task 自己改自己内改不外泄再看另一个场景async def worker(): trace_id.set(t-inner) async def main(): trace_id.set(t-outer) t asyncio.create_task(worker()) await t print(main sees:, trace_id.get()) # t-outerworker 里 set 的t-inner只作用在 worker 自己的上下文副本上不会污染 main 的上下文。这看起来很普通但它是隔离性的核心并发任务不会因为改了同一个 ContextVar 而互相干扰。这也是为什么可以放心在业务代码里直接 set而不会像全局变量那样出事。补充一点asyncio.gather、asyncio.wait_for这类组合 API也是基于当前上下文创建子任务的所以它们同样遵循“创建时快照”的规则。子任务如果再创建孙任务依然逐级捕获当前上下文形成一条完整链路。3.3 线程池asyncio.to_thread 自动带、run_in_executor 裸跑不带这是实际项目中很容易踩的细节。asyncio.to_thread会自动 copy 当前上下文然后在子线程里用ctx.run(func)执行。你从主协程 set 的值线程里能读到但线程里 set 的值不会回传。机制上很像“复印一份进去改”。import asyncio import contextvars info contextvars.ContextVar(info, defaultnone) def worker(): print(thread sees:, info.get()) info.set(changed-in-thread) async def main(): info.set(from-coroutine) await asyncio.to_thread(worker) print(main sees:, info.get()) asyncio.run(main()) # thread sees: from-coroutine # main sees: from-coroutine但如果你直接loop.run_in_executor(None, func)默认是没有这层包装的除非自己手动copy_context再 run。这个差异很容易坑人——同一个事件循环一个线程执行里有上下文另一个没有排查起来非常痛苦。如果一定要用run_in_executor可以自己包一层import contextvars def with_context(func, *args, **kwargs): ctx contextvars.copy_context() return ctx.run(func, *args, **kwargs)4. 实战五个高频场景的完整实现4.1 日志链路让 trace_id 贯穿异步调用日志里统一带一个 trace_id 是运维排查的基本功。用 contextvars 实现起来非常干净定义一个 TraceIdFilter每次写日志时从当前上下文读取 trace_id默认值给个短横线。import contextvars import logging import uuid TRACE_ID contextvars.ContextVar(trace_id, default-) class TraceIdFilter(logging.Filter): def filter(self, record: logging.LogRecord) - bool: record.trace_id TRACE_ID.get() return True logging.basicConfig( levellogging.INFO, format%(asctime)s [%(trace_id)s] %(name)s %(levelname)s %(message)s, ) logger logging.getLogger(app) # 请求入口 TRACE_ID.set(uuid.uuid4().hex[:8]) logger.info(request start) # 在任意协程里 logger.info(step inside coroutine)日志模块本身没有 trace_id 字段Filter 给它动态补上。配合前面的任务切换机制子任务里写日志也会自动带上同一个 trace_id不用每个协程手动传参数。这是日志链路的标准姿势。4.2 多租户系统认证后绑定租户多租户 SaaS 系统里最怕的就是查询时忘了带租户过滤条件。用 contextvars 可以把这个约束收敛到认证层tenant_id contextvars.ContextVar(tenant_id, defaultdefault) def authenticate(token: str): t decode(token)[tenant_id] return tenant_id.set(t) def query_ticket(): return db.query(ticket).filter_by(tenanttenant_id.get()).all()在中间件里调用authenticate拿到 token整个请求生命周期内不管代码在哪个协程里执行都能稳定取到当前租户。只要中间件设置位置靠前后续所有业务代码都不需要显式传递租户参数也少了很多“漏传租户导致数据越权”的问题。4.3 pytest 用例间隔离测试里用 contextvars 处理“每个用例独立会话”也很常见。autouse 的 fixture 可以确保每个用例开始前都初始化结束时清理。import pytest import contextvars session contextvars.ContextVar(session, defaultNone) pytest.fixture(autouseTrue) def reset_ctx(): token session.set(make_session()) # 每个用例独立 session yield session.reset(token)如果不做这个清理某个用例 set 了 session 而没重置下一个用例可能读到脏值。我在实际项目中真的遇过这种因为用例顺序依赖导致的随机失败最后就是靠这种 autouse fixture 解决的。4.4 在 FastAPI / ASGI 中间件里做请求级设置ASGI 中间件是设置上下文变量的最佳位置。请求进来时分配 request_id响应返回前 reset生命周期完全可控。from starlette.middleware.base import BaseHTTPMiddleware import uuid class RequestContextMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): request_id uuid.uuid4().hex[:12] token REQUEST_ID.set(request_id) try: response await call_next(request) return response finally: REQUEST_ID.reset(token)注意在 finally 里 reset。虽然 FastAPI 的请求处理本身也是基于 asyncio task但中间件往往横跨同一链路的多个阶段显式 reset 能让上下文状态更可控也避免某些复用路径上残留值。4.5 并发限流/连接绑定场景如果需要给每个并发协程绑定独立的连接 ID或者数据包需要携带当前处理者标识contextvars 同样好用CURRENT_CONN contextvars.ContextVar(current_conn, default0) async def process(client_id: int): conn_id_token CURRENT_CONN.set(client_id) try: for i in range(N): await asyncio.sleep(0.1) send_packet(CURRENT_CONN.get()) finally: CURRENT_CONN.reset(conn_id_token)这种场景如果用全局变量并发请求立刻串数据用 threading.local 又无法区分同一线程交错的任务contextvars 是正解。5. 高频踩坑清单上下文丢失、脏值与泄漏5.1 子任务读不到“晚设的值”最常见的问题是在主任务里create_task之后才 set 变量子任务永远读不到。async def main(): task asyncio.create_task(printer()) # 此时快照已经定了 MY_VAR.set(late-value) await task # printer 只能拿到 default如果你发现自己写了“先 create_task 后 set”的代码先想想子任务到底应该在哪个时间点拿到值。正确做法是把 set 放在create_task之前或者干脆把值作为参数传进去。5.2 忘记 reset长生命周期复用场景脏值在普通 HTTP 请求里asyncio 会自动为每个 Task 准备新 Context忘记 reset 影响可能不大。但当你手动复用Context.run、自定义 worker、或者把上下文绑定到可复用的连接对象上时不 reset 就会让下一次操作读到上一次的残留值。一个稳妥模式是 set 之后立即 try/finally reset不要图省事省略 finally。我写中间件时习惯先把 token 拿到再进 try最后在 finally 里 reset形成不会漏的固定结构。5.3 裸线程直接跑函数没有上下文ThreadPoolExecutor.submit(func)直接提交的函数默认不会自动接收当前上下文。你从协程切到线程池看到的可能全是默认值。解决办法有两个优先用asyncio.to_thread如果必须走裸线程池提交前手动 copy_context。from concurrent.futures import ThreadPoolExecutor import contextvars pool ThreadPoolExecutor(4) def run_with_ctx(func, *args, **kwargs): ctx contextvars.copy_context() return ctx.run(func, *args, **kwargs) future pool.submit(run_with_ctx, my_func, 1, 2)5.4 copy_context 是浅拷贝可变对象仍是共享的如果你在 ContextVar 里塞了 list、dict 这类可变对象copy_context()不会深拷贝它们。两个 Context 虽然各自保存了同一个列表的引用但修改这个列表的内容两边都会看到。data contextvars.ContextVar(data, default[]) async def worker(): data.get().append(changed) async def main(): data.set([]) task asyncio.create_task(worker()) await task print(data.get()) # [changed]所以放进 ContextVar 里的对象尽量遵循只读约定或者每次 set 时传入新的容器。否则你可能会“神奇地”在子任务里改动了父任务的列表还找不到原因。5.5 重复 reset 的 RuntimeError一个 token 只能 reset 一次。如果你在多个分支里都保存了 token并打算无条件 reset记得按逆序挨个 reset每个 token 只 reset 一次。实际编码中用一个 try/finally 包裹并在 finally 里 reset是足够安全的模式。提示Token 对象本身会暴露 var、old_value 属性但正常使用不需要读取这些字段。把它当作黑盒凭证是最合适的。我个人在项目里的习惯是能用参数显式传递的短链路不引入 contextvars一旦链路跨协程、跨线程或者分层很深就把 contextvars 收敛到中间件或装饰器里统一 set 和 reset永远不要在事件循环里手动把同一个 Context 用作多任务共享区。这三条帮我避开过不少奇怪的 bug。最后分享一个小技巧排查上下文问题时不要只盯着 get 的地方先去找 set 的地方和 create_task / to_thread 的调用时机。一路打日志确认“当前运行的到底是哪个任务、它的上下文从哪来的”通常比猜来得快。
