用 contextvars 在异步任务链中传递请求上下文避免全局污染

在同步 Web 应用里，把 request_id、当前登录用户、trace_id 这类信息塞进 threading.local 是很常见的做法。一旦迁到 asyncio 或 FastAPI/Starlette 这类异步栈上，这套写法立刻失效：协程在同一个线程里被事件循环交替调度，threading.local 里存的数据会互相串台，A 请求的 user_id

在同步 Web 应用里，把 request_id、当前登录用户、trace_id 这类信息塞进 threading.local 是很常见的做法。一旦迁到 asyncio 或 FastAPI/Starlette 这类异步栈上，这套写法立刻失效：协程在同一个线程里被事件循环交替调度，threading.local 里存的数据会互相串台，A 请求的 user_id 出现在 B 请求的日志里。另一种偷懒方案是搞一个模块级全局变量 CURRENT_USER，在并发场景下同样是灾难。

contextvars（PEP 567，Python 3.7+ 标准库）就是为这个问题设计的。它提供了一种「与执行上下文绑定」的变量：每个 asyncio.Task 在创建时会复制一份父上下文的快照，任务内部对变量的修改只对自身可见，任务之间互不影响。这篇文章从实际故障出发，讲清 ContextVar 的语义、在任务链中传递上下文的正确姿势、跨线程/跨进程的注意事项，以及几个真实踩坑报错的解法。

先看一个真实故障：threading.local 在协程里为什么会串数据

下面这段代码在很多老项目里都能找到影子，它的输出是不确定的：

import asyncio
import threading

_local = threading.local()


async def handle(user_id: str) -> None:
    _local.user_id = user_id
    await asyncio.sleep(0.01)          # 模拟一次数据库查询
    print(f"logged in as {_local.user_id}")   # 期望是自己的 user_id


async def main() -> None:
    await asyncio.gather(*(handle(f"u{i}") for i in range(5)))


asyncio.run(main())

实际输出（Python 3.11 实测）类似：

logged in as u4
logged in as u4
logged in as u4
logged in as u4
logged in as u4

原因很简单：asyncio.gather 创建的所有协程都跑在同一个线程里，threading.local 只有一个实例。sleep 让出控制权后，其他协程把 user_id 覆盖成最后一个值，等到恢复执行时读到的自然是最新写入的数据。这不是线程安全问题，而是「作用域粒度错了」——上下文应该跟随任务，而不是跟随线程。

把上面例子中的 threading.local 换成 ContextVar，输出立刻变成各自独立：

import asyncio
from contextvars import ContextVar

current_user: ContextVar[str] = ContextVar("current_user")


async def handle(user_id: str) -> None:
    current_user.set(user_id)
    await asyncio.sleep(0.01)
    print(f"logged in as {current_user.get()}")

注意 asyncio.gather 内部的实现是 loop.create_task()，而 create_task 会调用 contextvars.copy_context() 把创建时刻的上下文复制给新任务。因此每个子任务拿到的是一份独立快照，set() 只写进自己的那份。

ContextVar 的核心语义：set 返回 Token，reset 才能回滚

很多人第一次用 ContextVar 会犯一个错误：在中间件里 set() 完就完事了，请求结束不做清理，结果在复用上下文的场景（如后台任务、嵌套调用）里读到上一个请求的残值。ContextVar.set() 的返回值是一个 Token，这个 Token 记录了设置之前的值，通过 reset(token) 可以精确还原。

import asyncio
from contextvars import ContextVar

request_id: ContextVar[str] = ContextVar("request_id", default="unknown")


async def outer() -> None:
    token = request_id.set("req-001")
    print("inside outer:", request_id.get())      # req-001
    await inner()
    request_id.reset(token)
    print("after reset:", request_id.get())       # unknown


async def inner() -> None:
    print("inside inner:", request_id.get())      # req-001，沿调用链继承


asyncio.run(outer())

几个必须记住的规则：

  • 继承是「复制」而非「共享」：asyncio.Task、asyncio.create_task、Context.run 都会复制上下文。子任务里的 set 不会污染父上下文，父上下文的后续修改也不会回流到子任务。
  • 同一个协程内是共享的：普通函数调用、await 一个协程（而不是创建 Task）都在同一上下文内，变量可以正常传递。
  • 默认值只在从未 set 过时生效：如果声明了 default=...，get() 永不抛异常；不声明默认值且未 set 时，get() 抛 LookupError。
  • Token 不能跨上下文使用：在 A 上下文里拿到的 Token 拿到 B 上下文去 reset，会抛 ValueError: Token was created in a different Context。

也正因如此，set() 之后忘记 reset() 在「一个任务只处理一次请求」的模型下通常不出问题（任务结束上下文就销毁了），但在常驻协程、连接池消费者、以及手动 Context.run() 的代码里就会暴露。规范做法是成对使用，必要时包一个上下文管理器。

封装一个请求上下文对象，避免散落一堆 ContextVar

项目里通常不止一个字段。裸写五六个 ContextVar 会让代码到处 xxx.get()，也不好统一清理。更好的做法是用一个不可变 dataclass 承载所有字段，只暴露一个 ContextVar：

from __future__ import annotations

import uuid
from contextvars import ContextVar, Token
from dataclasses import dataclass, replace


@dataclass(frozen=True)
class RequestContext:
    request_id: str
    user_id: str | None = None
    tenant_id: str | None = None
    trace_id: str = ""

    def with_user(self, user_id: str) -> "RequestContext":
        return replace(self, user_id=user_id)


_ctx: ContextVar[RequestContext | None] = ContextVar("request_context", default=None)


def get_context() -> RequestContext:
    ctx = _ctx.get()
    if ctx is None:
        # 任务链外（如启动脚本）访问时给一个安全的兜底，避免 LookupError 打断业务流程
        return RequestContext(request_id="no-request", trace_id="no-trace")
    return ctx


def set_context(ctx: RequestContext) -> Token:
    return _ctx.set(ctx)


def bind_context(**kwargs) -> Token:
    """在现有上下文基础上覆盖部分字段，返回 Token 供 reset。"""
    return _ctx.set(replace(get_context(), **kwargs))


def reset_context(token: Token) -> None:
    _ctx.reset(token)

这样 get_context().tenant_id 在所有业务代码里都是同一套写法，中间件只需 set_context/reset_context 各一次。不可变 dataclass 保证了「改字段」必须走 bind_context，天然留下 Token，不会出现忘记回滚的路径。

一个配套的上下文管理器，能让 with 块自动清理，减少手写 token 的机会：

from contextlib import contextmanager
from typing import Iterator


@contextmanager
def request_context(request_id: str, **kwargs) -> Iterator[RequestContext]:
    ctx = RequestContext(request_id=request_id, **kwargs)
    token = set_context(ctx)
    try:
        yield ctx
    finally:
        reset_context(token)

在 ASGI 中间件里注入上下文，并让它顺着 await 链流下去

FastAPI/Starlette 的中间件本身就是 async 函数，天然处于请求的上下文里，适合做注入点。关键点有两个：一是 call_next 内部会创建新任务，依赖 asyncio.Task 的上下文复制才能把值带进路由函数；二是必须在 finally 里 reset，防止 BaseHTTPMiddleware 复用上下文时出问题。

import uuid

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

from ctx import RequestContext, set_context, reset_context, get_context

app = FastAPI()


class RequestContextMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = request.headers.get("x-request-id") or uuid.uuid4().hex
        trace_id = request.headers.get("traceparent", "")
        ctx = RequestContext(request_id=request_id, trace_id=trace_id)
        token = set_context(ctx)
        try:
            response = await call_next(request)
        finally:
            reset_context(token)
        response.headers["x-request-id"] = request_id
        return response


app.add_middleware(RequestContextMiddleware)


@app.get("/orders/{order_id}")
async def read_order(order_id: str):
    ctx = get_context()
    # 这里无论是直接读，还是再多层 await，拿到的都是本请求自己的 request_id
    return {"order_id": order_id, "request_id": ctx.request_id}

启动后用 curl 验证：

uvicorn main:app --port 8000
curl -s -H 'x-request-id: abc123' http://127.0.0.1:8000/orders/42
# {"order_id":"42","request_id":"abc123"}

再压一下并发，确认不串台：

for i in $(seq 1 20); do
  curl -s -H "x-request-id: req-$i" http://127.0.0.1:8000/orders/$i &
done
wait

每个响应体里的 request_id 与请求头一致，说明上下文隔离生效。如果要给日志自动带上 request_id，可以给 logging 加一个 Filter，在 filter() 里读 get_context()，Formatter 里用 %(request_id)s 占位：

import logging

from ctx import get_context


class ContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        ctx = get_context()
        record.request_id = ctx.request_id
        record.user_id = ctx.user_id or "-"
        return True


handler = logging.StreamHandler()
handler.addFilter(ContextFilter())
handler.setFormatter(logging.Formatter(
    "%(asctime)s %(levelname)s [%(request_id)s] [%(user_id)s] %(name)s: %(message)s"
))
logging.basicConfig(level=logging.INFO, handlers=[handler])

注意：logging 自身的 Logger 在 Python 3.12 之前并不会自动携带创建时的上下文，所以 Filter 里实时读取 get_context() 是更稳的写法，也别指望 logging.setLogRecordFactory 能跨任务生效。

把上下文带进线程池、后台任务和子进程

异步代码里最容易出事的不是主链路，而是「跳出去」的那几处。

1. run_in_executor 不会自动传递上下文

loop.run_in_executor 把函数丢进线程池，线程池里的线程没有当前任务的上下文。直接在里面 get_context() 拿到的是兜底值。

import asyncio
import contextvars
from concurrent.futures import ThreadPoolExecutor

from ctx import get_context, request_context


def blocking_query() -> str:
    # 在 executor 线程里执行
    return get_context().request_id


async def main() -> None:
    async with request_context("req-777"):
        loop = asyncio.get_running_loop()
        with ThreadPoolExecutor(max_workers=4) as pool:
            # 错误写法：上下文丢失，得到 no-request
            bad = await loop.run_in_executor(pool, blocking_query)

            # 正确写法：用 copy_context 包一层
            good = await loop.run_in_executor(
                pool, contextvars.copy_context().run, blocking_query
            )
        print("bad =", bad, "| good =", good)


asyncio.run(main())

输出：

bad = no-request | good = req-777

contextvars.copy_context().run 把当前上下文的快照带进线程里执行，是 to_thread / run_in_executor 场景的标准补丁。Python 3.9+ 的 asyncio.to_thread 内部已经调用了 contextvars.copy_context()，所以直接用它就不用手动包：

from asyncio import to_thread

async def handler() -> str:
    return await to_thread(blocking_query)   # 自动继承上下文

2. 自己 create_task 的守护任务需要显式复制

asyncio.create_task 会自动复制创建时刻的上下文，这一点与 ensure_future/TaskGroup 一致。但如果任务是「先创建、后延迟执行」，且你期望它看到执行时刻的最新上下文，就得手动 copy_context().run 或重新 set。反过来，如果你希望后台任务完全不受请求上下文影响（比如全局清理任务），在任务入口显式清空反而更安全。

3. 多进程与 Celery 不共享内存

ContextVar 只存在于单个进程内。multiprocessing 子进程、ProcessPoolExecutor、Celery worker 都拿不到父进程的上下文。正确做法是把 request_id/tenant_id 作为显式参数塞进消息体，消费端进入任务后重新 set_context(...)。Celery 可以在任务基类里统一注入：

from celery import Task

from ctx import RequestContext, set_context, reset_context


class ContextTask(Task):
    def __call__(self, *args, **kwargs):
        meta = self.request.headers or {}
        ctx = RequestContext(
            request_id=meta.get("request_id", "celery-task"),
            tenant_id=meta.get("tenant_id"),
        )
        token = set_context(ctx)
        try:
            return super().__call__(*args, **kwargs)
        finally:
            reset_context(token)

发送端用 apply_async(headers={"request_id": get_context().request_id}) 把链路串起来，这样日志里的 request_id 才能从 HTTP 一路贯穿到异步任务。

四个真实报错与排查思路

下面这些异常都是在引入 contextvars 之后项目里实际出现过的，按频率排序。

报错信息 触发原因 解决办法
LookupError: <ContextVar name='request_context' at 0x...> 声明时没有 default，且在未 set 的上下文里调用了 get()（常见于启动钩子、单元测试、executor 线程） 要么给 ContextVar(..., default=None) 并在读取处兜底，要么在入口统一 set_context
ValueError: Token was created in a different Context 在任务 A 中 set() 得到 Token，却在任务 B 或 Context.run 出来的新上下文里 reset(token) set 和 reset 必须在同一执行上下文内成对出现，不要跨 await create_task 边界传递 Token
RuntimeError: cannot enter context: <Context object at ...> is already entered 同一个 contextvars.Context 对象被并发 run()，例如把 copy_context() 的结果缓存后复用 每次 run 前重新 copy_context()，Context 对象不能并发进入（Python 3.11+ 部分场景放宽，但不要依赖）
日志里 request_id 为 - 或 no-request，但接口返回值正常 日志来自线程池/Celery/子进程，上下文没带过去 线程池用 copy_context().run 或 asyncio.to_thread；跨进程把 id 放进消息体并在消费端重建上下文

另外一个非异常但很常见的坑：在 @app.on_event("startup") 或 lifespan 里 set_context 之后，想让这个值对后续所有请求生效——这是做不到的。启动阶段设置的上下文只属于事件循环启动时的那份快照，每个新任务都会复制它，但请求中间件里的 set 不会回流到别处。全局配置请用普通模块变量或 app.state，ContextVar 专管「请求级」数据。

还有一点值得强调：ContextVar 不解决线程安全问题。如果在上下文里存一个可变 dict 并在多个任务间共享引用，照样会互相污染。要么存不可变对象（如前文的 frozen dataclass），要么每次 bind_context 生成新副本。

小结

contextvars 的定位非常明确：把「请求级/任务级」数据从线程或全局变量里挪出来，绑定到执行上下文。asyncio.Task 创建时自动复制上下文，这是它在异步栈上优于 threading.local 的根本原因。落地时有四条纪律：用不可变 dataclass 聚合字段、set 与 reset(token) 成对出现、跨线程用 copy_context().run（或直接用 asyncio.to_thread）、跨进程把标识放进消息体并在消费端重建上下文。做到这几点，日志串台、用户身份错乱、租户数据越权这类问题基本可以根除。

FAQ

ContextVar 和 threading.local 能混用吗？

可以共存但不要混着表达同一个语义。已经用 ContextVar 表示请求上下文后，不要再在代码里加一层 threading.local 缓存，否则线程池场景下会出现两套数据不一致。统一走 ContextVar，在需要跨线程时复制上下文即可。

为什么在 FastAPI 的依赖注入里 get_context() 有时拿不到值？

如果依赖函数是 async def，它与路由函数在同一任务上下文内，能拿到中间件 set 的值。如果依赖写成 def（同步函数），Starlette 会把它放到线程池执行，此时上下文可能丢失。解决办法是改成 async def，或者在同步依赖内部改为通过参数显式传递所需字段。

ContextVar 的 default 设成可变对象安全吗？

不安全。默认值在多个上下文之间是同一个对象引用，未 set 的地方 get() 出来的是同一份数据，修改会互相影响。默认值只放 None、字符串这类不可变值，需要容器时用 default=None 加显式创建。

能不能用 ContextVar 做全局限流计数器或缓存？

不建议。每个任务复制上下文后各自持有独立副本，increment 只影响自己那份，统计不到全局。跨请求共享的可变状态应该用 asyncio.Lock 保护的普通对象、app.state，或者外部存储（Redis）。

Python 3.7 以下怎么办？

contextvars 是 3.7 加入标准库的。3.6 需要安装 PyPI 上的 contextvars 回移包；3.5 及更早没有可靠方案，只能显式传参。目前 FastAPI 等框架已要求 3.8+，实际项目中不需要考虑回移。

测试里怎么验证上下文隔离？

用 pytest-asyncio 写两个并发任务，各自 set 不同 request_id，在中间 await asyncio.sleep(0) 让出控制权，再断言读取值等于自己写入的值。这个用例能同时覆盖「串台」和「未清理」两类回归。