在线 -
搜索 AIGC编程网
← 返回题目列表

异步代码里怎么传递上下文？为什么不能用 threading.local？

中等 第 18 / 27 题 更新于 2026/08/01
contextvars上下文传递trace_idasyncio

简化版

threading.local 在异步代码里会「串」——因为一个线程里跑着成千上万个协程，它们共享同一个线程本地存储。当协程 A 在 await 处让出控制权、协程 B 接着跑并覆盖了同一个 threading.local 的值，A 恢复后读到的就是 B 的数据（典型症状：日志里的 request_id 张冠李戴、A 用户的请求带上了 B 用户的租户 ID）。正确的工具是 contextvars（Python 3.7+，PEP 567）：var = ContextVar("request_id")，用 var.set(v) / var.get() 读写。它的关键机制是——每个 asyncio.Task 在创建时会「复制一份当前上下文的快照」，之后这个任务里对 ContextVar 的修改只影响它自己，既不会污染父任务、也不会被兄弟任务看见。所以「一个请求一条链路」的上下文能自动跟着协程走，穿过任意深的 await 调用栈，不需要一层层传参。四个必须记住的点：① var.set() 返回一个 Token，配合 var.reset(token) 才能正确恢复上一层的值（在中间件、装饰器里尤其重要）；② ContextVar 必须在模块层创建（不要在函数里每次新建，否则每次都是不同的变量）；③ asyncio.to_thread 会自动传递上下文（它内部用 contextvars.copy_context()），但手动 loop.run_in_executor 不会；④ 子任务的修改不会回传给父任务——想「向上传递」结果必须用返回值或可变对象。核心记忆：协程共享线程，所以 threading.local 会串；用 contextvars，Task 创建时拷贝快照、修改只影响自己。

详细版

两者对比：

维度threading.localcontextvars.ContextVar
隔离单位线程上下文（每个 Task 一份快照）
异步下❌ 会串（协程共享线程）✅ 正确隔离
多线程下✅✅（每个线程有独立的默认上下文）
传播方式不传播创建 Task 时复制快照
恢复上一层手动保存Token + reset()
版本一直有3.7+
import asyncio, contextvars, threading, logging

# ① ★ContextVar 必须在模块层创建★
request_id = contextvars.ContextVar("request_id", default="-")
tenant = contextvars.ContextVar("tenant")          # 不给 default → get() 会抛 LookupError

# ② 基本读写
token = request_id.set("req-123")     # ★返回 Token★
print(request_id.get())                # 'req-123'
request_id.reset(token)                # ★恢复到 set 之前的值★
print(request_id.get())                # '-'（default）

# ③ ★threading.local 在异步里会串★
_local = threading.local()
async def bad(name):
    _local.name = name
    await asyncio.sleep(0.1)           # ★让出控制权，别的协程会覆盖 _local.name★
    print(f"期望 {name}，实际 {_local.name}")   # ★大概率不一致★

# ④ ✓ contextvars 正确隔离
async def good(name):
    request_id.set(name)
    await asyncio.sleep(0.1)
    print(f"期望 {name}，实际 {request_id.get()}")   # ✅ 永远一致

async def main():
    await asyncio.gather(*(good(f"req-{i}") for i in range(3)))
    # ★每个 Task 有自己的上下文快照，互不干扰★

# ⑤ ★快照语义：创建 Task 时拷贝，之后互不影响★
async def child():
    print("子任务看到:", request_id.get())   # ★看到创建时刻父任务的值★
    request_id.set("changed-in-child")       # ★只改自己的副本★

async def parent():
    request_id.set("parent-value")
    await asyncio.create_task(child())
    print("父任务仍然是:", request_id.get())  # ★'parent-value'——子任务改不到父任务★

# ⑥ ★典型用途：日志里自动带上 trace_id★
class ContextFilter(logging.Filter):
    def filter(self, record):
        record.request_id = request_id.get()   # ★从上下文取★
        return True
handler = logging.StreamHandler()
handler.addFilter(ContextFilter())
handler.setFormatter(logging.Formatter("%(asctime)s [%(request_id)s] %(message)s"))

# ⑦ 中间件里注入（FastAPI/Starlette 风格）
async def trace_middleware(request, call_next):
    token = request_id.set(request.headers.get("X-Request-ID", new_id()))
    try:
        return await call_next(request)
    finally:
        request_id.reset(token)          # ★必须 reset★

# ⑧ ★to_thread 会传递上下文，run_in_executor 不会★
def sync_work():
    return request_id.get()              # ✅ to_thread 里能读到

async def demo():
    request_id.set("abc")
    print(await asyncio.to_thread(sync_work))            # 'abc' ✅
    loop = asyncio.get_running_loop()
    print(await loop.run_in_executor(None, sync_work))   # ★'-'（拿不到）★
    # ✓ 修法：ctx = contextvars.copy_context()
    #        await loop.run_in_executor(None, ctx.run, sync_work)

⚠️ 三个必须理解的机制：① threading.local 的隔离单位是「线程」，而 asyncio 是「一个线程跑成千上万个协程」——所有协程共享同一份线程本地存储，谁最后写谁说了算。协程 A 在 await 处让出后，B 覆盖了值，A 恢复时读到的就是 B 的——这类 bug 在低并发时不复现、上了量就到处串号，而且没有任何异常，只是数据错了。② ContextVar 的传播靠「Task 创建时的快照」：asyncio.create_task()（以及 gather、TaskGroup 内部）会调用 contextvars.copy_context() 拷贝当前上下文，任务体在这份副本里运行。所以父 → 子是「创建那一刻」的值（之后父的修改子看不到）、子 → 父完全不回传。想让子任务的结果影响父任务，只能靠返回值或共享的可变对象。③ set() 必须配 reset(token)：ContextVar 没有「作用域自动退出」的机制，在中间件、装饰器里设置后如果不 reset，这个值会一直留在当前上下文里——在复用的任务或线程池的工作线程里就会泄漏到下一个请求（to_thread 的工作线程虽然每次都是新的 Context 副本，但同一个 Task 内的后续代码会受影响）。

完整版教学

一、为什么 threading.local 在异步里失效

threading.local 的模型：
  线程 1 ──► 自己的一份存储
  线程 2 ──► 自己的一份存储
  → 隔离单位 = ★线程★

asyncio 的模型：
  ★一个线程★ ──► 协程 A、协程 B、协程 C … （成千上万个）
  → 所有协程★共享同一个线程★ → ★共享同一份 threading.local★

出错的时间线（★把这个讲清楚就答对了一半★）：
  t0  协程 A：_local.request_id = "req-A"
  t1  协程 A：await db.query()        ← ★让出控制权★
  t2  协程 B：_local.request_id = "req-B"   ← ★覆盖了同一份存储★
  t3  协程 B：await ...               ← 让出
  t4  协程 A：恢复执行
      log.info(_local.request_id)     → ★"req-B"！错的★

  ★ 现象特征：
    - 低并发时不复现（协程之间没有交错）
    - 高并发时到处串号
    - ★没有任何异常★，只是数据错了（最难查的那种 bug）
    - 日志里 request_id 张冠李戴、A 用户看到 B 用户的租户配置

★ 为什么不能"每个协程一个线程"来绕过：
  asyncio 的价值就是"一个线程支撑十万并发"
  → 十万个线程 = 800GB 虚拟内存 + 调度崩溃
  → 所以必须有一个"比线程更细"的隔离单位 → ★这就是 Context★

对比三种隔离单位：
  ┌──────────────┬────────────────┬──────────────────────────┐
  │ 机制          │ 隔离单位        │ 适用                      │
  ├──────────────┼────────────────┼──────────────────────────┤
  │ 全局变量      │ 进程            │ 配置（只读）              │
  │ threading.local│ ★线程★        │ 多线程同步代码            │
  │ ★contextvars★│ ★Context（协程）│ ★异步代码（也兼容多线程）★│
  └──────────────┴────────────────┴──────────────────────────┘
  ★ contextvars 在多线程下同样正确（每个线程有自己的默认 Context）
  → 所以★新代码统一用 contextvars★，不用再区分同步/异步

threading.local 失效的原因非常直接：它的隔离单位是「线程」，而 asyncio 是「一个线程跑成千上万个协程」——所有协程共享同一份线程本地存储。那条时间线说明了一切：协程 A 写入后在 await 处让出，协程 B 覆盖了同一份存储，A 恢复时读到的就是 B 的值。这类 bug 的特征是「低并发不复现、高并发到处串号、且没有任何异常」——日志里 request_id 张冠李戴、A 用户看到 B 用户的租户配置，是最难排查的一类问题。也不能用「每个协程一个线程」来绕过——asyncio 的全部价值就在于「一个线程支撑十万并发」，所以必须有一个比线程更细的隔离单位，这就是 Context。顺带记住：contextvars 在多线程下同样正确（每个线程有自己的默认上下文），所以新代码统一用它，不用再区分同步还是异步。

二、contextvars 的三个概念

★ ContextVar：上下文变量（"键"）
  var = ContextVar("name", default=值)      # ★必须在模块层创建★
  var.set(v)   → 返回 Token
  var.get()    → 取值；没有值且没有 default → ★LookupError★
  var.get(默认值) → 取不到时返回这个默认值（★不会抛异常★）

  ★ 为什么必须在模块层创建：
    ContextVar 对象本身就是"键"，每次 ContextVar("x") 都是★一个全新的键★
    ✗ def f():
          v = ContextVar("x")      # ★每次调用都是新变量，永远读不到别人设的值★
    ✓ V = ContextVar("x")          # 模块层，全局唯一

★ Context：上下文（"一份键值映射的快照"）
  ctx = contextvars.copy_context()          # ★复制当前所有 ContextVar 的值★
  ctx.run(func, *args)                      # ★在这份副本里运行 func★
  → func 里对 ContextVar 的修改★只影响 ctx★，不影响调用方
  dict(ctx)                                 # 可以查看里面有什么

★ Token：撤销凭证
  token = var.set("new")
  ...
  var.reset(token)     # ★恢复到 set 之前的状态★（包括"之前没有值"这个状态）
  ★ token 只能用一次，且只能在★设置它的那个 Context★ 里 reset
  ★ 用错会抛 ValueError: Token was created in a different Context

三者关系：
  ┌────────────────────────────────────────────────┐
  │  Context（一份快照）                             │
  │    ├─ ContextVar("request_id") → "req-123"      │
  │    ├─ ContextVar("tenant")     → "acme"         │
  │    └─ ContextVar("user")       → User(...)      │
  └────────────────────────────────────────────────┘
  Token 记录"某个 var 在 set 之前的状态"，用于精确回滚

典型的作用域封装（★推荐写法★）：
  from contextlib import contextmanager
  @contextmanager
  def use_request_id(value):
      token = request_id.set(value)
      try:
          yield
      finally:
          request_id.reset(token)      # ★异常路径也恢复★

  with use_request_id("req-1"):
      ...                              # 这段代码里 request_id 是 req-1
  # 出来之后自动恢复

contextvars 有三个概念。ContextVar 是「键」——必须在模块层创建，因为 ContextVar("x") 每次调用都产生一个全新的键，在函数里新建的话永远读不到别人设的值；get() 在没有值且没有 default 时会抛 LookupError（用 get(默认值) 可避免）。Context 是「一份键值映射的快照」——copy_context() 复制当前所有变量的值，ctx.run(func) 让 func 在这份副本里运行、它的修改不影响调用方。Token 是「撤销凭证」——var.set() 返回它，var.reset(token) 能精确恢复到 set 之前的状态（包括「之前根本没有值」这个状态）；注意 token 只能用一次，且只能在设置它的那个 Context 里 reset，用错会抛 ValueError。实践中推荐用 @contextmanager 把「set + reset」封装成一个作用域，保证异常路径也能恢复。

三、asyncio 里的传播规则

★ 核心规则：创建 Task 时复制一份上下文快照★

  asyncio.create_task(coro)
    → 内部执行 contextvars.copy_context()
    → 任务体在这份★副本★里运行

  由此推出四条行为：
  ① ★父 → 子：传递「创建那一刻」的值★
     request_id.set("A")
     t = asyncio.create_task(child())     # ★child 看到 "A"★
     request_id.set("B")                  # ★之后改的，child 看不到★
  ② ★子 → 父：完全不回传★
     child 里 request_id.set("C") → ★父任务仍然是 "B"★
  ③ ★兄弟之间：互相隔离★
     gather 出来的多个任务各有各的快照
  ④ ★同一个协程内（没有 create_task）：共享上下文★
     await some_coro()   ← ★直接 await，不创建 Task → 共享当前上下文★
     → 所以 some_coro 里的 set() ★会影响调用方★

  ★ 这条区别非常重要：
    await coro()               → ★同一个上下文★（修改可见）
    await create_task(coro())  → ★新快照★（修改不可见）
    → gather(coro1(), coro2()) 内部会把裸协程包装成 Task → ★各自快照★

★ 各种 API 的上下文传播行为（★速查表★）：
  asyncio.create_task / ensure_future     ★复制快照★
  asyncio.gather(裸协程)                   ★每个包装成 Task → 各自快照★
  TaskGroup.create_task                    ★复制快照★
  直接 await 协程                          ★共享当前上下文★
  loop.call_soon / call_later              ★复制快照★（可传 context= 指定）
  ★asyncio.to_thread★                     ★复制并在线程里 ctx.run★ ✅
  ★loop.run_in_executor★                  ★不传递★ ❌（要自己 copy_context）
  concurrent.futures 的 submit             ★不传递★ ❌
  异步生成器                                共享创建时的上下文（★注意 aclose 时机★）

★ 一个容易踩的坑：run_in_executor 不传上下文
  ✗ await loop.run_in_executor(None, sync_func)      # sync_func 里读不到
  ✓ ctx = contextvars.copy_context()
    await loop.run_in_executor(None, ctx.run, sync_func)
  ✓ 或直接用 asyncio.to_thread（★内部已经处理★）

★ 另一个坑：回调里的上下文
  loop.call_later(1, callback)      # ★默认复制"注册时"的上下文★
  → callback 里能读到注册时的 request_id ✅
  但如果你用 functools.partial 手动包装、或存到别处稍后调用 → 要自己传 Context

asyncio 的传播规则只有一条核心：创建 Task 时复制一份上下文快照（create_task 内部调用 copy_context()）。由此推出四条行为：父 → 子传递「创建那一刻」的值（之后父的修改子看不到）、子 → 父完全不回传、兄弟任务互相隔离、以及同一个协程内（直接 await，不创建 Task）共享上下文。最后这条区别非常重要：await coro() 是共享上下文（里面的 set() 会影响调用方），而 await create_task(coro()) 是新快照（修改不可见）——gather(coro1(), coro2()) 内部会把裸协程包装成 Task，所以各自快照。速查表里要特别记住两个：asyncio.to_thread 会自动传递上下文（内部用 ctx.run），而 loop.run_in_executor 不会（需要自己 copy_context() 后传 ctx.run）；concurrent.futures 的 submit 同样不传递。

四、典型用途与实践模板

★ 用途一：请求链路追踪（最经典）★
  # context.py
  request_id: ContextVar[str] = ContextVar("request_id", default="-")
  user_id: ContextVar[str | None] = ContextVar("user_id", default=None)

  # 中间件（FastAPI/Starlette）
  @app.middleware("http")
  async def trace(request, call_next):
      rid = request.headers.get("X-Request-ID") or uuid4().hex
      token = request_id.set(rid)
      try:
          resp = await call_next(request)
          resp.headers["X-Request-ID"] = rid
          return resp
      finally:
          request_id.reset(token)        # ★必须★

  # 日志自动注入
  class ContextFilter(logging.Filter):
      def filter(self, record):
          record.request_id = request_id.get()
          record.user_id = user_id.get() or "-"
          return True
  → ★业务代码里 logger.info("xxx") 就自动带上了 trace 信息，不用层层传参★

★ 用途二：多租户 / 权限上下文★
  tenant: ContextVar[str] = ContextVar("tenant")
  # ORM 层自动加过滤条件
  def build_query(model):
      return select(model).where(model.tenant_id == tenant.get())
  ★ 好处：不用每个函数都传 tenant 参数
  ★ 风险：★隐式依赖★——忘记设置时 LookupError，或更糟地用了默认值查了全量数据
    ✓ 对策：不给 default，让"忘记设置"直接报错（★fail fast★）

★ 用途三：数据库会话 / 事务★
  session: ContextVar[AsyncSession] = ContextVar("session")
  # 请求开始时 set，结束时 reset + close
  ★ SQLAlchemy 的 async_scoped_session 就是用 contextvars 实现的
    （scopefunc=asyncio.current_task 或 contextvars）

★ 用途四：分布式追踪（OpenTelemetry）★
  OTel 的 Context 传播底层就是 contextvars
  → span 自动嵌套、trace_id 自动传递到下游调用

★ 完整的作用域管理模板：
  from contextlib import asynccontextmanager

  @asynccontextmanager
  async def request_scope(rid: str, uid: str | None = None):
      t1 = request_id.set(rid)
      t2 = user_id.set(uid)
      try:
          yield
      finally:
          user_id.reset(t2)             # ★注意 reset 顺序（后设的先 reset）★
          request_id.reset(t1)

  async with request_scope("req-1", "u-42"):
      await handle()

★ 反模式：
  ✗ 用 contextvars 传递"业务参数"（应该显式传参）
    → 上下文是给"横切关注点"用的（日志、追踪、租户、认证）
    → 业务逻辑靠隐式上下文会变得极难测试和理解
  ✗ 在库的 API 里依赖调用方设置 ContextVar（★隐式契约★）
    → 库应该提供显式参数，contextvars 只作为可选的便利

contextvars 的典型用途都是横切关注点：请求链路追踪（中间件里 set 一次，日志 Filter 自动注入，业务代码不用层层传参）、多租户/权限上下文（ORM 层自动加过滤条件）、数据库会话（SQLAlchemy 的 async_scoped_session 就是基于它）、以及分布式追踪（OpenTelemetry 的 Context 传播底层就是 contextvars）。实践模板的要点是用 @asynccontextmanager 封装「set + reset」，注意 reset 的顺序应该和 set 相反（后设的先 reset）。多租户场景有个重要的设计取舍：不要给 tenant 设 default——让「忘记设置」直接抛 LookupError 快速失败，好过用一个默认值悄悄查了全量数据。最后要警惕反模式：别用 contextvars 传递业务参数（那应该显式传参，隐式上下文会让代码极难测试），库的公开 API 也不应该依赖调用方设置 ContextVar（隐式契约）。

五、常见陷阱

★ 陷阱一：忘记 reset，值泄漏到后续代码★
  async def handler(request):
      request_id.set(request.id)      # ★没有 reset★
      await do_work()
  → 在同一个 Task 里后续执行的代码仍然看得到这个值
  → 在复用 Task 的框架里（某些自定义 worker）会泄漏到下一个请求
  ✓ 永远用 try/finally + reset，或封装成上下文管理器

★ 陷阱二：在函数里创建 ContextVar★
  ✗ def setup():
        v = ContextVar("request_id")    # ★每次调用都是新的键★
        v.set("x")
  → 别处 get() 永远读不到（因为不是同一个 ContextVar 对象）
  ✓ 模块层定义，全局唯一

★ 陷阱三：以为子任务的修改会影响父任务★
  async def worker():
      result_var.set(compute())        # ★父任务读不到★
  await asyncio.gather(worker(), worker())
  print(result_var.get())              # ★还是旧值★
  ✓ 用返回值：results = await asyncio.gather(...)
  ✓ 或用共享的可变对象（list/dict）——但那就要考虑并发写

★ 陷阱四：run_in_executor 丢上下文★
  ✓ 用 asyncio.to_thread（★自动处理★）
  ✓ 或 ctx = copy_context(); loop.run_in_executor(None, ctx.run, fn)

★ 陷阱五：异步生成器的上下文★
  async def gen():
      print(request_id.get())          # ★哪个值？★
      yield 1
  → 异步生成器在"每次 __anext__ 被调用时"运行，
    上下文取决于★调用方当时的上下文★（不是创建时的）
  ★ 3.12+ 对异步生成器的上下文处理有改进；跨上下文使用生成器要格外小心
  ✓ 保守做法：把需要的上下文值在创建生成器时★取出来作为参数★

★ 陷阱六：线程池 worker 复用★
  ThreadPoolExecutor 的工作线程会被复用
  → 如果在里面 set 了 ContextVar 而没有 reset
  → ★下一个任务可能读到上一个任务的值★
  （用 ctx.run 时每次是新副本，相对安全；但直接 set 到线程默认上下文就会残留）

★ 陷阱七：性能★
  ContextVar.get() 很快（★内部有缓存★，约几十纳秒）
  set() 稍慢（要创建新的 Context 映射）
  copy_context() 的成本随变量数量增长
  → ★正常使用完全不用担心★；但别在超高频循环里反复 set

★ 陷阱八：和 threading.local 混用★
  同一个项目里两套机制 → 有的地方串有的地方不串 → ★极难排查★
  ✓ ★统一用 contextvars★（它在多线程下同样正确）

八个陷阱里最常见的是忘记 reset（值会泄漏到同一 Task 的后续代码，在复用 Task 的框架里甚至泄漏到下一个请求）和在函数里创建 ContextVar（每次都是新键，别处永远读不到）。「子任务的修改不会影响父任务」也经常被误解——想拿到结果只能靠返回值。异步生成器的上下文要特别小心：它在每次 __anext__ 被调用时运行，上下文取决于调用方当时的上下文而非创建时的，保守做法是把需要的值在创建时取出来作为参数。线程池 worker 复用也是隐患：在工作线程里 set 了不 reset，下一个任务可能读到残留值。性能上不用担心——get() 内部有缓存、约几十纳秒，只是别在超高频循环里反复 set。最后一条纪律：统一用 contextvars，不要和 threading.local 混用（两套机制并存会导致「有的地方串有的地方不串」，极难排查）。

六、和其他方案的对比与选型

★ 四种"传递上下文"的方式：
  ① ★显式传参★
     async def handle(req, *, trace_id): ...
     ✓ 最清晰、最好测试、无隐式依赖
     ✗ 调用链深时到处都要加参数（"参数污染"）
     → ★业务参数用这个★

  ② ★contextvars★
     ✓ 穿透任意深的调用栈、协程安全、多线程也正确
     ✓ 库/框架能自动获取（日志、追踪、ORM）
     ✗ ★隐式依赖★——看函数签名不知道它依赖什么
     ✗ 测试时要记得设置上下文
     → ★横切关注点用这个★（trace_id、租户、认证、session）

  ③ ★threading.local★
     ✗ ★异步下会串★
     → 只在"纯多线程同步代码"里用；新代码建议直接上 contextvars

  ④ ★对象属性 / 依赖注入★
     class Handler:
         def __init__(self, ctx): self.ctx = ctx
     ✓ 显式、可测试
     → 适合"一个请求一个 Handler 实例"的架构

★ 选型准则：
  ┌────────────────────────────┬────────────────────┐
  │ 场景                        │ 推荐                │
  ├────────────────────────────┼────────────────────┤
  │ 业务参数（订单号、金额）      │ ★显式传参★         │
  │ trace_id / request_id       │ ★contextvars★     │
  │ 租户 / 当前用户              │ ★contextvars★     │
  │ 数据库 session / 事务        │ contextvars 或 DI  │
  │ 配置（只读、全局一致）        │ 模块级常量          │
  │ 纯同步多线程的线程私有数据    │ contextvars（也可以）│
  └────────────────────────────┴────────────────────┘

★ 框架里的实际情况：
  FastAPI/Starlette：ContextVar 是官方推荐的请求上下文方案
                     （Starlette 的 request 本身用 ★scope 字典★传递）
  Flask：g / request 是 ★werkzeug 的 LocalProxy★
         → 早期基于 threading.local，★新版已改用 contextvars★以支持 async
  Django：async 视图下的上下文也迁移到了 contextvars（asgiref 内部大量使用）
  ★asgiref 的 sync_to_async / async_to_sync 会正确传递 contextvars★
  OpenTelemetry / Sentry：全部基于 contextvars 做 trace 传播

★ 测试时的注意：
  单测里如果被测函数依赖 ContextVar，要先 set：
  def test_x():
      token = request_id.set("test-1")
      try:
          assert build_log() == "[test-1] ..."
      finally:
          request_id.reset(token)
  ✓ 或用 pytest fixture 封装
  ★ 这也是"隐式依赖"的代价——所以只用于横切关注点

传递上下文有四种方式，各有定位：显式传参最清晰最好测试，业务参数应该用它；contextvars 能穿透任意深的调用栈且协程安全，横切关注点（trace_id、租户、认证、session）用它，代价是隐式依赖（看签名不知道依赖什么，测试时要记得设置）；threading.local 在异步下会串，新代码直接上 contextvars 即可；依赖注入适合「一个请求一个 Handler 实例」的架构。框架层面的现状值得知道：FastAPI/Starlette 官方推荐用 ContextVar 做请求上下文；Flask 的 g/request 早期基于 threading.local、新版已改用 contextvars 以支持 async；asgiref 的 sync_to_async/async_to_sync 会正确传递 contextvars；OpenTelemetry 和 Sentry 的 trace 传播全部基于它。测试时要记得先 set 再断言、并在 finally 里 reset——这正是「隐式依赖」的代价，所以它只适合横切关注点。

记忆钩子：「★threading.local 在异步里会串，因为它的隔离单位是『线程』，而 asyncio 是『一个线程跑成千上万个协程』★——协程 A 在 await 处让出、协程 B 覆盖同一份存储、A 恢复后读到的就是 B 的值；症状是★低并发不复现、高并发到处串号、且没有任何异常★（日志 request_id 张冠李戴、A 用户看到 B 租户的数据）。正解是 ★contextvars（3.7+，PEP 567）★，三个概念：★ContextVar 是键（必须在模块层创建★，因为每次 ContextVar(‘x’) 都是全新的键；get() 无值无 default 会 LookupError）、★Context 是一份快照★（copy_context() + ctx.run(fn)）、★Token 是撤销凭证★（set() 返回它，reset(token) 精确恢复到 set 之前，含『之前没有值』这个状态）。★核心传播规则只有一条：创建 Task 时复制一份上下文快照★，由此推出：父→子传『创建那一刻』的值、★子→父完全不回传★（想拿结果只能靠返回值）、兄弟任务互相隔离，而★直接 await 协程是共享上下文★（里面的 set 会影响调用方）——gather(裸协程) 内部会包装成 Task 所以各自快照。速查两个易错：★asyncio.to_thread 会传递上下文（内部 ctx.run），loop.run_in_executor 不会★（要自己 copy_context 后传 ctx.run），concurrent.futures 的 submit 同样不传。★必须 set 配 reset★（用 @contextmanager 封装保证异常路径也恢复），否则值会泄漏到后续代码、在复用 Task 或线程池 worker 里泄漏到下一个请求。用途都是★横切关注点★：trace_id 注入日志、多租户（★不给 default 让忘记设置 fail fast★）、DB session（SQLAlchemy 的 async_scoped_session 就基于它）、OpenTelemetry 的 trace 传播；★业务参数仍应显式传参★。最后：contextvars 在多线程下同样正确，★新代码统一用它、别和 threading.local 混用★（Flask 新版的 g/request 也已从 threading.local 迁移过来）。」

七、常见误区与追问

  • 误区：threading.local 在 asyncio 里也能用，反正每个请求逻辑上是独立的。 会串数据。threading.local 的隔离单位是线程，而 asyncio 的模型是「一个线程上跑成千上万个协程」——它们共享同一份线程本地存储。典型时间线：协程 A 写入 _local.request_id = "A" 后在 await 处让出控制权，协程 B 接着运行并写入 "B"，A 恢复后读到的就是 "B"。这类 bug 的可怕之处在于：低并发时协程之间没有交错，完全不复现；上了量之后到处串号，而且不抛任何异常，只是数据错了——日志里 request_id 张冠李戴、A 用户的请求带上了 B 用户的租户 ID、审计记录归错人。正确工具是 contextvars，它的隔离单位比线程更细。
  • 误区：在函数或类的 __init__ 里创建 ContextVar 更灵活。 必须在模块层创建。ContextVar 对象本身就是「键」——每次调用 ContextVar("request_id") 都会产生一个全新的、互不相同的键，即使名字字符串一样也毫无关系（那个名字只用于 repr 和调试）。所以在函数里创建的话，你 set 到的是这次调用创建的那个变量，而别处 get 的是另一个变量，永远读不到值（抛 LookupError 或拿到 default）。正确做法是在模块顶层定义成全局常量（通常放在专门的 context.py 里），像 request_id: ContextVar[str] = ContextVar("request_id", default="-") 这样，全项目引用同一个对象。
  • 误区：在子任务里 set 了值，父任务或兄弟任务也能读到。 读不到。asyncio.create_task() 在创建任务时会调用 contextvars.copy_context() 复制一份当前上下文的快照，任务体在这份副本里运行——所以子任务的 set() 只改自己的副本，既不回传给父任务，也不影响兄弟任务（这正是隔离的意义）。同样地，父任务在创建子任务之后的修改，子任务也看不到（它拿的是「创建那一刻」的快照）。想让子任务的结果传回父任务，只能用返回值（results = await asyncio.gather(...)）或共享的可变对象（但那要自己考虑并发写）。这里还有个关键区别：直接 await coro() 不创建 Task，因此共享当前上下文，里面的 set() 是会影响调用方的。
  • 误区：set() 之后不用管，函数返回时上下文会自动恢复。 不会——ContextVar 没有「作用域自动退出」的机制，set() 的效果会一直保留在当前 Context 里，直到被再次 set 或 reset。后果分两种：① 同一个 Task 内的后续代码会读到这个残留值（比如中间件设置后没清，响应处理阶段仍然带着它）；② 更严重的是复用场景——某些框架会复用 Task 或工作线程，在线程池的 worker 里 set 了不 reset，下一个任务可能读到上一个任务的值。正确做法是 set() 必须配 reset(token)，并用 try/finally 或封装成 @contextmanager/@asynccontextmanager 保证异常路径也能恢复；多个变量时 reset 的顺序应与 set 相反。
  • 误区：loop.run_in_executor 和 asyncio.to_thread 一样，都会把上下文带到线程里。 只有 asyncio.to_thread 会——它内部实现就是 ctx = contextvars.copy_context() 然后 loop.run_in_executor(None, ctx.run, func, *args)，所以线程里的同步函数能正常读到 ContextVar。而直接调用 loop.run_in_executor(None, func) 不做任何上下文传递，函数在工作线程的默认上下文里运行，get() 只会拿到 default 或抛 LookupError——表现为「日志里的 trace_id 到了线程池任务里就丢了」。修法有两个：优先用 asyncio.to_thread（3.9+），或者自己写 ctx = copy_context() 再 run_in_executor(None, ctx.run, func)。同理，concurrent.futures 的 submit()、以及你自己 threading.Thread(target=...) 起的线程都不会传递上下文。
  • 追问：contextvars 和 threading.local 能不能混用？在多线程同步代码里该用哪个？ 不建议混用——同一个项目里两套机制并存会导致「有的地方串、有的地方不串」，排查时极其痛苦。新代码统一用 contextvars 即可，因为它在纯多线程同步代码里同样正确：每个线程有自己的默认 Context，一个线程里的 set() 不会影响另一个线程。这样做还有两个额外好处：① 代码同时兼容同步和异步（将来把某个函数改成 async def 不用重写上下文机制）；② 生态一致——asgiref 的 sync_to_async/async_to_sync 会正确传递 contextvars，OpenTelemetry、Sentry 等库也都基于它。事实上 Flask 的 g 和 request 早期基于 threading.local，新版已经迁移到 contextvars 正是为了支持 async 视图。唯一需要注意的是：在线程池里用时仍要正确 reset（因为 worker 线程会被复用）。
  • 追问：为什么 copy_context() 是「快照」而不是「引用」？这样设计有什么好处？ 因为快照语义保证了任务之间的隔离：如果是引用共享，那么一个任务修改 request_id 就会影响所有其他任务——这就退化成了 threading.local 的问题。快照的具体行为是：创建 Task 时把当前所有 ContextVar 的值复制成一份不可变映射（内部用 HAMT 持久化数据结构实现，所以复制的成本很低，是 O(1) 的结构共享而不是深拷贝），之后任务对变量的 set() 会在自己这份映射上产生新版本，不影响原来的。这个设计带来三个好处：① 天然的父子隔离（子任务的修改不泄漏出去）；② 值本身仍然是共享的对象（复制的只是「键→值」的映射，不会深拷贝你的对象）；③ 性能可接受（get() 内部还有缓存，约几十纳秒）。代价是「子→父」的传递必须走返回值，但这反而让数据流更清晰。
  • 追问：异步生成器里的 ContextVar 行为是怎样的？ 这是最容易出错的边界。异步生成器的函数体不是在创建时执行的，而是在每次 __anext__() 被调用时才恢复执行——所以它读到的上下文取决于调用方当时所处的上下文，而不是生成器创建时的上下文。如果一个生成器被跨任务、跨上下文地消费（比如在 A 任务里创建、传给 B 任务去迭代），里面的 ContextVar.get() 会读到 B 的值，这通常不是你想要的。另外还有清理时机问题：异步生成器的 aclose() 可能由事件循环在完全不同的上下文里触发（asyncio 有专门的 async generator 终结机制），此时 finally 块里读上下文同样不可靠。保守做法是：在创建生成器时就把需要的上下文值取出来，作为参数传进去（gen(trace_id=request_id.get())），而不是在生成器体内 get()。Python 3.12 起对异步生成器的上下文处理有所改进，但「显式取值传参」始终是最稳妥的。

八、加强记忆

threading.local 在异步里会串，因为它的隔离单位是「线程」，而 asyncio 是「一个线程跑成千上万个协程」——协程 A 在 await 处让出、协程 B 覆盖同一份存储、A 恢复后读到的就是 B 的值；症状是低并发不复现、高并发到处串号、且没有任何异常（日志 request_id 张冠李戴、A 用户看到 B 租户的数据）。正解是 contextvars（3.7+，PEP 567），三个概念：ContextVar 是键（必须在模块层创建，因为每次 ContextVar("x") 都是全新的键；get() 在无值无 default 时抛 LookupError）、Context 是一份快照（copy_context() + ctx.run(fn)，内部用 HAMT 做结构共享所以复制很廉价）、Token 是撤销凭证（set() 返回它，reset(token) 精确恢复到 set 之前，包括「之前没有值」这个状态）。核心传播规则只有一条：创建 Task 时复制一份上下文快照，由此推出：父 → 子传「创建那一刻」的值、子 → 父完全不回传（想拿结果只能靠返回值）、兄弟任务互相隔离；而直接 await 协程是共享上下文（里面的 set() 会影响调用方）——gather(裸协程) 内部会包装成 Task 所以各自快照。速查两个易错点：asyncio.to_thread 会传递上下文（内部就是 ctx.run），而 loop.run_in_executor 不会（需自己 copy_context() 后传 ctx.run），concurrent.futures.submit 同样不传。set() 必须配 reset()（用 @contextmanager 封装保证异常路径也恢复，多个变量时 reset 顺序与 set 相反），否则值会泄漏到后续代码、在复用 Task 或线程池 worker 里泄漏到下一个请求。用途都是横切关注点：trace_id 注入日志、多租户（不给 default 让「忘记设置」fail fast）、DB session（SQLAlchemy 的 async_scoped_session 就基于它）、OpenTelemetry 的 trace 传播；业务参数仍应显式传参。异步生成器是最易错的边界——它读到的是消费时的上下文而非创建时的，保守做法是创建时取值传参。最后：contextvars 在多线程下同样正确，新代码统一用它、别和 threading.local 混用（Flask 新版的 g/request 也已迁移过来）。