Python contextvars：任务与线程之间的上下文传播边界

请求编号在协程日志里正常，搬进线程池就变成了 -，并不一定是日志配置漏了一行。ContextVar 的值属于当前上下文；把函数交给另一种执行入口时，必须确认那个入口有没有携带上下文，以及携带的是哪个时点的值。

还有一种更容易误判的现象：子任务把请求编号改成了新值，父任务读到的却仍是旧值。这恰好说明隔离正在生效。上下文传播不是一块让所有执行者共同修改的全局字典。

下面固定 Python 3.12，只用标准库，用事件确定任务已经创建、何时读取，不靠随机延时碰时序。目标是把四条路径放到同一张可运行的证据链里：创建任务、等待 to_thread、使用线程池，以及显式复制上下文。

先看任务创建的那个时点

asyncio.create_task() 没有传 context 时，会复制调用者当前的上下文。之后父任务再次赋值，并不会把已经创建的子任务一起改掉。子任务也可以在自己的上下文里继续赋值，父任务不用跟着接收那份修改。

保存为 context_routes.py：

import asyncio
import contextvars
from concurrent.futures import ThreadPoolExecutor

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


def thread_read_and_change():
    before = request_id.get()
    token = request_id.set("thread-local-change")
    try:
        return before, request_id.get()
    finally:
        request_id.reset(token)


async def task_reader(gate):
    await gate.wait()
    before = request_id.get()
    token = request_id.set("child-change")
    try:
        await asyncio.sleep(0)
        return before, request_id.get()
    finally:
        request_id.reset(token)


async def main():
    root_token = request_id.set("request-A")
    try:
        gate = asyncio.Event()
        child = asyncio.create_task(task_reader(gate))
        request_id.set("request-B")
        gate.set()
        child_result = await child
        assert child_result == ("request-A", "child-change")
        assert request_id.get() == "request-B"
        print("task:", child_result, "parent:", request_id.get())

        # to_thread 返回协程。到它开始执行时，才复制当前上下文。
        pending = asyncio.to_thread(thread_read_and_change)
        request_id.set("request-C")
        thread_result = await pending
        assert thread_result == ("request-C", "thread-local-change")
        assert request_id.get() == "request-C"
        print("to_thread:", thread_result, "parent:", request_id.get())

        loop = asyncio.get_running_loop()
        # 独立新建的线程池，worker 没有继承这里的 request-C。
        with ThreadPoolExecutor(max_workers=1) as pool:
            plain = await loop.run_in_executor(pool, request_id.get)
            assert plain == "-"

            ctx = contextvars.copy_context()
            copied = await loop.run_in_executor(
                pool, ctx.run, thread_read_and_change
            )
            assert copied == ("request-C", "thread-local-change")
            after = await loop.run_in_executor(pool, request_id.get)
            assert after == "-"
            print("executor:", plain, "copied:", copied, "after:", after)
    finally:
        request_id.reset(root_token)

    assert request_id.get() == "-"
    print("root restored:", request_id.get())


if __name__ == "__main__":
    asyncio.run(main())

执行：

python3 --version
python3 context_routes.py

Python 3.12 下，输出中的关键部分是：

task: ('request-A', 'child-change') parent: request-B
to_thread: ('request-C', 'thread-local-change') parent: request-C
executor: - copied: ('request-C', 'thread-local-change') after: -
root restored: -

第一行没有竞争条件。子任务在 request-A 时创建，却要等事件放行才读取；父任务已经设置了 request-B，子任务依然拿到 A。决定默认上下文副本的是任务创建，而不是子协程第一次调用 get()。

to_thread 的复制动作，发生在它真正执行时

上面的第二段特意把创建协程与等待分成两行。调用 asyncio.to_thread(...) 得到协程对象，并不会仅凭这一行就立即完成线程提交。随后将值改成 C，再 await pending，线程看到的是 C。

如果希望固定在当前时点，可以立即把它安排成任务：

async def capture_when_scheduled():
    token = request_id.set("scheduled-value")
    try:
        task = asyncio.create_task(asyncio.to_thread(request_id.get))
        request_id.set("later-value")
        assert await task == "scheduled-value"
    finally:
        request_id.reset(token)

这里不是给 to_thread 增加了某种新的传播模式。外层 Task 先复制了上下文，后续 to_thread 在那份上下文中执行，再把相应值带进线程。排查时应把协程对象、Task、线程工作项这三个动作分开看，不能统称“启动异步函数”。

to_thread 还解决不了线程函数的强制停止。外部协程被取消，已经运行的同步函数仍可能继续执行。上下文正确，只代表它能读到适当的标识，不代表它已经获得了请求生命周期的全部控制能力。

自建线程池需要明确的传播封装

实验为 run_in_executor 使用独立的新线程池，因此 - 的来源清楚。不要让测试依赖某个复用 worker 之前是否执行过会修改上下文的业务。

业务确实要求传播时，可以在提交边界复制，再让 ctx.run 包住实际调用。每次提交使用新副本，尤其不能把同一个 Context 实例同时拿给多个线程进入；一个上下文不能在多个地方同时处于已进入状态。

ctx.run 返回后会退出那份上下文，所以最后一次直接读取 worker 又得到 -。这比在线程函数开头设置一个值、结尾忘记恢复可靠得多。长寿命 worker 会处理许多请求，任何未清理的状态都可能影响下一项工作。

但跨进程任务不能照搬这个封装。进程池参数必须服从进程间传输规则，请求编号等必要信息应作为明确数据传递，由执行端建立自己的上下文。

直接 await 一个协程，并没有创建隔离层

一个普通的 await read_value() 会让协程继续在当前任务中执行。若这个协程调用 request_id.set() 且没有恢复，调用者随后就会读到修改后的值。不能把“函数写成 async”当成上下文隔离的证据；上面第一段隔开父子的动作，是创建了新的 Task。

这也是入口清理仍然需要 token 的原因。恢复不等于统一设置成默认值：入口之前可能已经存在一份外层上下文，退出应该还给外层原值。把出口写死成 set("-")，会破坏中间件嵌套、测试复用等情况。token 记录的是对应那次设置之前的绑定，且必须在正确上下文中使用，不能带到另一任务里替父任务重置。

记录日志时还要看读取发生在哪里。工作者创建日志记录后再交给监听线程，监听线程自己的当前上下文可能没有请求编号。需要的标识最好在产生日志事件时成为记录字段，而不是等另一个线程输出时才读取。传播和事件封装分属两个边界，修好前一个并不会自动修好后一个。

上下文副本不深拷贝里面的对象

把字符串换成字典，问题会立即变样。假设把 {"request_id": "A"} 放进一个 ContextVar，复制上下文以后，两份上下文最初仍可以引用同一个字典。对其中一份执行 set(新字典)，只改变该上下文的绑定；对原字典执行原地修改，却会让其他持有者看见变化。

因此，请求编号、租户编号这类简单元数据适合保存为不可变值。若要携带复杂配置，应让被引用对象保持只读，或者在确实需要独立修改的边界建立新的对象。copy_context() 名字中的“复制”，不能成为深层状态隔离的证据。

同样不宜把数据库连接、可变事务对象或业务授权结果一股脑藏进上下文。调用栈上看不见依赖，生命周期又可能被后台任务延长，排障反而更难。上下文更适合减少日志元数据的重复传参；真正决定业务行为的输入，通常仍值得显式表达。

入口设置一次，出口在 finally 中用原 token 恢复；创建后台任务时决定是否应继承；跨线程时检查所用入口的承诺。把这几个位置固定下来，请求编号丢失和串线才有具体的排查路线，而不是靠在更多函数里补 set() 试运气。