Python contextvars 在异步任务中怎么传请求上下文：Task 边界、线程池与日志关联

异步接口里经常要把 request_id、租户标识或当前用户带进下游任务，直接把它们塞进每层函数参数很快会变成“参数污染”。Python 的 contextvars 适合保存这类请求级状态：同一个事件循环中的 asyncio.Task 会继承创建时的上下文，不同 Task 之间又保持隔离；但进入线程池后不能想当然地认为上下文仍然存在，必要时要用 copy_context() 明确传递。

先把请求上下文放进 ContextVar

请求入口只设置一次，业务函数通过 .get() 读取。默认值建议明确写出，这样后台任务或测试漏掉初始化时会得到可识别结果，而不是在日志里出现难追踪的异常。

from contextvars import ContextVar

request_id = ContextVar("request_id", default="-")

def current_request_id() -> str:
    return request_id.get()

token = request_id.set("A-100")
try:
    print(current_request_id())  # A-100
finally:
    request_id.reset(token)

set() 返回的 Token 记录了修改前的状态。不要只在函数尾部再次 set(default)，那样会破坏调用方原本的上下文；reset(token) 才能按嵌套顺序恢复。

asyncio Task 为什么能保持隔离

创建 Task 时，事件循环会自动把当前上下文复制给新生成的任务。多个异步任务从不同入口给同名变量赋值，也不会出现互相覆盖的情况，刚好适配多请求并行处理、同一请求拆分多个下游异步分支这类场景。

import asyncio
from contextvars import ContextVar

request_id = ContextVar("request_id", default="-")

async def worker(name: str, value: str) -> None:
    request_id.set(value)
    await asyncio.sleep(0)
    print(name, request_id.get())

async def main() -> None:
    await asyncio.gather(
        asyncio.create_task(worker("A", "A-100")),
        asyncio.create_task(worker("B", "B-200")),
    )

asyncio.run(main())

验收重点不是输出顺序，而是每个任务都打印自己的值。若把 request_id 换成普通模块变量，两个协程在交错运行时就会产生覆盖，日志关联会随机串线。

进入线程池时，先确认上下文传播方式

asyncio.to_thread() 是较省心的入口，它会把调用点的当前上下文带到线程函数。若项目仍使用 loop.run_in_executor() 或自建 ThreadPoolExecutor，则应显式复制上下文。

import asyncio
import contextvars
from concurrent.futures import ThreadPoolExecutor

request_id = contextvars.ContextVar("request_id", default="-")

def read_context() -> str:
    return request_id.get()

async def main() -> None:
    request_id.set("A-100")
    async_value = await asyncio.to_thread(read_context)

    ctx = contextvars.copy_context()
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor(max_workers=1) as pool:
        fixed_value = await loop.run_in_executor(pool, ctx.run, read_context)

    print(async_value, fixed_value)  # A-100 A-100

手动线程池的典型故障是日志格式里有 request_id=-，但业务主协程明明已经设置了请求号。这个现象不代表 ContextVar 丢值，而是线程拥有自己的上下文；把 ctx.run 作为线程任务入口即可补上边界。

运维排查按这张边界表走

停机与异常路径不要遗忘 reset

中间件、测试夹具和重试包装器都可能嵌套设置同一个变量。最小可靠写法是“设置—执行—恢复”三段式。即使下游抛出异常，外层上下文也不会被污染；取消任务时同样会进入 finally。

async def handle(request_id_value: str) -> None:
    token = request_id.set(request_id_value)
    try:
        await call_downstream()
    finally:
        request_id.reset(token)

如果需要把上下文交给一个长期运行的后台队列，别把整个请求上下文无限期挂在任务对象上。更稳妥的做法是把必要的 request_id 或租户标识作为消息字段传入，消费端重新设置并在处理完成后恢复。

常见问题

ContextVar 是不是线程安全的全局变量？

不是。它是按上下文维度隔离的变量，每个线程和每个异步 Task 持有的上下文相互独立，不会互相干扰。

create_task 创建后再 set，子任务会看到新值吗？

正常不会出现这类异常串值问题。Task 默认会捕获创建那一刻的上下文快照，只要在任务创建前就设好对应变量，或是手动构造明确的上下文传入即可。

copy_context 会复制出一个完全独立的上下文吗？

它复制当前上下文映射，后续在复制上下文里修改不会直接改掉调用方的变量；可用 ctx.run() 把目标函数放进去执行。

日志库怎么读取 request_id？

日志过滤器或适配器在记录时调用 request_id.get() 即可。跨线程场景先验证上下文传播，再排查日志配置本身。

收尾检查

把 ContextVar 当作请求上下文容器，而不是共享状态容器：Task 边界依赖创建时捕获，线程边界要确认传播方式，嵌套修改必须用 Token 恢复。上线前用两个并行 Task、一次 to_thread 和一次手动线程池调用打印同一字段，基本就能定位关联链路是否完整。