在 asyncio 服务里给每个请求打上 trace_id，并在该请求派生的所有协程、线程池任务、第三方库日志里自动带上它，看起来是个小需求，实际踩坑率极高。最常见的做法是把 trace_id 塞进 logging.LoggerAdapter 或者一路当函数参数往下传：前者在 asyncio.create_task 之后就会丢失上下文，后者会污染每一层函数签名，跨库调用时根本传不过去。更隐蔽的问题是，用 threading.local 存 trace_id 在单线程事件循环里“碰巧能用”，一旦接入 run_in_executor 或者 FastAPI 的同步依赖，就会读到其他请求的脏值，日志串号，排查线上问题时把人带进沟里。
本文用一个可运行的完整示例，说明如何用标准库 contextvars 建立请求级上下文、如何让 asyncio 任务自动继承它、如何把 trace_id 注入到 logging 的 Formatter 里、以及如何跨线程池与跨进程（如 httpx 调用下游）继续传递。所有代码基于 Python 3.11 验证，contextvars 自 3.7 起为标准库，无第三方依赖也能跑通核心逻辑。
为什么 threading.local 和参数传递都不可靠
先把问题边界划清楚。一个 Web 请求从进入到返回，可能经历：
- 事件循环里的一串
await，中间被await asyncio.sleep()切走，同一线程上还跑着几十个别的请求； asyncio.create_task()派生的后台任务（写审计日志、发消息队列）；loop.run_in_executor()丢到线程池的阻塞调用（比如同步的数据库驱动、requests）；- 第三方库自己打的日志，你没法改它的函数签名。
threading.local 的问题在于：它按线程隔离，而 asyncio 是单线程多协程，几十个请求共享一个线程，谁最后写入谁覆盖。下面这段代码能直接在 3.11 上复现串号：
import asyncio
import threading
_local = threading.local()
async def handle(req_id: str, delay: float) -> None:
_local.req_id = req_id
await asyncio.sleep(delay) # 让出控制权，另一个协程写入 _local
# 读到的可能是别人的 req_id
print(f"expect={req_id} actual={getattr(_local, 'req_id', None)}")
async def main() -> None:
await asyncio.gather(handle("A", 0.2), handle("B", 0.1))
asyncio.run(main())
# 输出：
# expect=B actual=B
# expect=A actual=B <-- 串号
把 trace_id 当参数一路传，则在“第三方库打日志”这个场景直接失效：logging.getLogger("httpx").info(...) 内部没有你的参数。所以需要一个“隐式、协程安全、可跨任务继承”的存储，这正是 contextvars.ContextVar 的设计目标。
ContextVar 的语义：copy、继承与显式设置
ContextVar 的值存在 contextvars.Context 对象里，而不是存在线程或协程对象上。关键规则只有三条，记住就不会踩坑：
- 每个 asyncio Task 创建时复制当前 Context。
asyncio.create_task(coro)内部调用contextvars.copy_context()，任务内对ContextVar.set()的修改只影响该任务自己的副本，任务结束即丢弃。 await不会切换 Context。同一任务内的多次await共享同一个 Context，因此在入口处 set 一次，后续所有await都能读到。- 线程池不会自动继承。
run_in_executor把函数丢到另一个线程，那个线程用的是它自己的空 Context，必须手动ctx.run(fn)或copy_context()传过去。
还有一个常被忽略的细节：ContextVar.set() 返回一个 Token，用 var.reset(token) 可回滚。在请求入口用 try/finally 配对 set/reset 是最稳的写法，尤其在进程内复用的中间件里，避免上一个请求的值泄漏到下一个。下面是最小验证，证明 Task 之间互不干扰：
import asyncio
from contextvars import ContextVar
trace_id: ContextVar[str] = ContextVar("trace_id", default="-")
async def worker(name: str) -> None:
print(f"{name} start trace={trace_id.get()}")
await asyncio.sleep(0.05)
print(f"{name} end trace={trace_id.get()}")
async def request(name: str) -> None:
trace_id.set(name)
await asyncio.gather(worker(f"{name}-w1"), worker(f"{name}-w2"))
async def main() -> None:
# 注意：直接在 main 里 set 再 gather，两个 request 会共享同一个 Context
await asyncio.gather(
asyncio.create_task(request("req-A")),
asyncio.create_task(request("req-B")),
)
asyncio.run(main())
输出里每个 worker 都稳定打印自己所属请求的 trace，不会互相污染。相反，如果不套 create_task，直接 await request("req-A") 与 await request("req-B") 串行执行当然也没问题，但并发时若两个协程跑在同一个 Context 里（例如都从 main 直接 gather），后 set 的会覆盖前一个——这一点在写测试和框架中间件时务必分清。
日志自动注入：用 logging.Filter 挂到 Handler 上
logging 模块本身不会读 ContextVar。标准做法是写一个 Filter，在 LogRecord 进入 Formatter 之前把 trace_id 塞进 record 的 __dict__，然后 Formatter 用 %(trace_id)s 引用。必须挂在 Handler 上而不是 Logger 上：Logger 级别的 Filter 只对直接调用该 logger 的记录生效，子 logger（app.db、httpx 等）经 propagate 冒泡上来时会绕过它；挂到 Handler 上则所有冒泡到该 Handler 的记录都会被处理。
import logging
from contextvars import ContextVar
trace_id_var: ContextVar[str] = ContextVar("trace_id", default="-")
class TraceIdFilter(logging.Filter):
"""把当前 ContextVar 里的 trace_id 注入 LogRecord。"""
def filter(self, record: logging.LogRecord) -> bool:
record.trace_id = trace_id_var.get()
return True
def setup_logging(level: int = logging.INFO) -> None:
fmt = "%(asctime)s %(levelname)-7s [%(trace_id)s] %(name)s: %(message)s"
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(fmt, datefmt="%Y-%m-%d %H:%M:%S"))
handler.addFilter(TraceIdFilter())
root = logging.getLogger()
root.setLevel(level)
root.handlers.clear()
root.addHandler(handler)
两个易错点。第一，如果某个第三方库或旧代码手动构造 LogRecord 并绕过 Filter（例如某些库直接调用 handler.emit），格式化时会出现 KeyError: 'trace_id'，因此要么在 Filter 里保证 set，要么给 Formatter 一个兜底：
class SafeFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
if not hasattr(record, "trace_id"):
record.trace_id = "-"
return super().format(record)
第二，record.trace_id 是直接写属性，重复添加 Filter 不会报错但会多一次赋值；在测试里反复调用 setup_logging 时记得先 root.handlers.clear()，否则日志会重复输出两次，很多“日志刷屏但内容一样”的 bug 都源于此。
跨线程池与下游调用：copy_context 与显式透传
同步阻塞代码被丢进线程池时，需要手动搬运 Context。标准库提供了两种等价写法，推荐第一种，携带业务参数更直观：
import asyncio
import contextvars
import logging
from concurrent.futures import ThreadPoolExecutor
logger = logging.getLogger(__name__)
_pool = ThreadPoolExecutor(max_workers=4, thread_name_prefix="blocking")
def blocking_db_call(sql: str) -> int:
# 这里的 trace_id_var.get() 能读到，因为外层用 ctx.run 搬运了 Context
logger.info("executing: %s", sql)
return 1
async def run_blocking(sql: str) -> int:
ctx = contextvars.copy_context()
loop = asyncio.get_running_loop()
return await loop.run_in_executor(_pool, ctx.run, blocking_db_call, sql)
注意 loop.run_in_executor(pool, ctx.run, fn, *args) 的参数顺序，是把 ctx.run 当被调函数、fn 当它的第一个参数传进去，等价于在线程里执行 ctx.run(blocking_db_call, sql)。Context 对象不可重复进入（同一时刻只能在一个线程里 run），所以每次调用都要重新 copy_context()，不要缓存一个全局 ctx 复用，否则会撞上 RuntimeError: cannot enter context: ... is already entered。
跨进程/跨 HTTP 下游则没有自动传播，只能显式透传。如果下游也是 Python 服务，通常走 HTTP Header（如 X-Trace-Id 或 W3C 的 traceparent）。用 httpx 的 event_hooks 可以少写重复代码：
import httpx
async def inject_trace_header(request: httpx.Request) -> None:
request.headers["X-Trace-Id"] = trace_id_var.get()
client = httpx.AsyncClient(
base_url="http://inventory.internal",
timeout=httpx.Timeout(3.0, connect=1.0),
event_hooks={"request": [inject_trace_header]},
)
下游收到后，在入口中间件里 trace_id_var.set(header_value) 即可续接链路。如果没有上游头，就生成一个新的：trace_id_var.set(uuid.uuid4().hex)。截断长度和字符集要和下游约定一致，uuid4().hex 是 32 位十六进制，兼容性最好。
完整可运行示例：一个 asyncio 服务骨架
下面把所有片段拼成一个文件，python demo.py 直接运行，输出里每条日志都带请求自己的 trace_id，后台任务与线程池调用同样不串号。它不依赖任何 Web 框架，方便你先验证语义再迁移到 FastAPI/Starlette。
import asyncio
import contextvars
import logging
import uuid
from concurrent.futures import ThreadPoolExecutor
trace_id_var: ContextVar[str] = ContextVar("trace_id", default="-")
class TraceIdFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
record.trace_id = trace_id_var.get()
return True
def setup_logging() -> None:
fmt = "%(asctime)s %(levelname)-7s [%(trace_id)s] %(name)s: %(message)s"
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(fmt, datefmt="%H:%M:%S"))
handler.addFilter(TraceIdFilter())
root = logging.getLogger()
root.setLevel(logging.INFO)
root.handlers.clear()
root.addHandler(handler)
logger = logging.getLogger("demo")
_pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="blocking")
def slow_sync_io(sql: str) -> str:
logger.info("thread=%s sql=%s", "pool", sql) # 在池线程里打日志
return "ok"
async def run_sync(sql: str) -> str:
ctx = contextvars.copy_context()
loop = asyncio.get_running_loop()
return await loop.run_in_executor(_pool, ctx.run, slow_sync_io, sql)
async def audit(msg: str) -> None:
await asyncio.sleep(0.03)
logger.info("audit: %s", msg)
async def handle_request(path: str) -> None:
token = trace_id_var.set(uuid.uuid4().hex[:12])
try:
logger.info("enter %s", path)
await asyncio.sleep(0.05)
result = await run_sync(f"select 1 /* {path} */")
logger.info("db result=%s", result)
asyncio.create_task(audit(f"done {path}"))
finally:
trace_id_var.reset(token)
async def main() -> None:
setup_logging()
await asyncio.gather(*(handle_request(f"/api/item/{i}") for i in range(3)))
await asyncio.sleep(0.1) # 等 fire-and-forget 的 audit 任务跑完
if __name__ == "__main__":
asyncio.run(main())
典型输出（每次 trace_id 不同，但同一请求内一致）：
10:21:04 INFO [8f3a1c9d7b02] demo: enter /api/item/0
10:21:04 INFO [b41e77a0c5e9] demo: enter /api/item/1
10:21:04 INFO [d90c2f6a83b1] demo: enter /api/item/2
10:21:04 INFO [8f3a1c9d7b02] demo: thread=pool sql=select 1 /* /api/item/0 */
10:21:04 INFO [b41e77a0c5e9] demo: thread=pool sql=select 1 /* /api/item/1 */
10:21:04 INFO [d90c2f6a83b1] demo: thread=pool sql=select 1 /* /api/item/2 */
10:21:04 INFO [8f3a1c9d7b02] demo: db result=ok
10:21:04 INFO [8f3a1c9d7b02] demo: audit: done /api/item/0
audit 是 fire-and-forget 的 create_task，它能带上 trace_id，正是因为 Task 创建时复制了当前 Context；run_sync 能带上，是因为 ctx.run 把 Context 搬进了池线程。这两处正是线上最常见的“日志丢 trace”位置。若你把入口那行 try/finally 里的 reset 删掉，在长生命周期进程里（比如自建的 worker 循环复用线程）就会看到上一个请求的 id 粘到下一个请求上。
如果接入 FastAPI，把 handle_request 的逻辑放进一个 BaseHTTPMiddleware，或者更推荐用 ASGI 中间件（避免 BaseHTTPMiddleware 对 Context 的已知干扰），并在响应头里回写 X-Trace-Id 便于前端和网关串联。也有人直接用 Starlette 的 contextvars 友好写法，但注意 BaseHTTPMiddleware 会为每个请求创建 Task，set 发生在哪一侧决定了后续能否读到，遇到“中间件里 set 了但视图读不到”的报错，先确认这一点。
小结
请求级 trace_id 的正确姿势是：用 ContextVar 做存储，在请求入口 set 并用 Token 在 finally 里 reset；日志侧用挂在 Handler 上的 Filter 注入 record.trace_id，Formatter 用 %(trace_id)s 引用；asyncio.create_task 自动继承无需处理，run_in_executor 必须用 copy_context().run 搬运；跨进程靠 HTTP Header 显式透传。三条规则覆盖 95% 的场景，剩下 5% 是第三方库自己缓存 logger 或手动构造 LogRecord，用兜底 Formatter 和最小复现逐个击破。
FAQ
为什么我在中间件里 set 了 trace_id，视图函数里 get 到的还是默认值 “-”？
九成是 Task 边界问题。若用 FastAPI/Starlette 的 BaseHTTPMiddleware，它对下游请求的处理可能运行在另一个 Task 的 Context 副本里，你在中间件主体里 set 的值不会回流到视图所在的 Context。解决办法是改用纯 ASGI 中间件（直接 await self.app(scope, receive, send)），或把 set 放进视图依赖（Depends）里。排查时在该处打印 id(contextvars.copy_context()) 和 trace_id_var.get() 对照即可。另一个可能是你用了 logger.addFilter(...) 而非 handler.addFilter(...)，子 logger 冒泡时不经过 Logger 级 Filter。
日志里出现 KeyError: ‘trace_id’ 怎么办？
说明某条 LogRecord 到达 Formatter 时没有 trace_id 属性。通常是有代码绕过了 Handler 上的 Filter，例如直接用 logging.makeLogRecord({...}) 后手工 emit，或某个库自建了 Handler。最省事的修复是改用 SafeFormatter（见上文），在 format 里对缺失属性补默认值；根治则是统一走 trace_id_var.set() 并在 Filter 中赋值。
run_in_executor 抛 RuntimeError: cannot enter context 是什么原因？
你把同一个 contextvars.Context 对象并发地传给了多个线程。Context 不允许同时进入两次。每次提交任务前重新 contextvars.copy_context()，不要把它存在全局变量或闭包里复用。另外 ThreadPoolExecutor 线程复用，若你在池线程里手动 set 而不 reset，下一个任务会读到上一条的 trace_id，同样要用 try/finally 配对。
能用 logging.LoggerAdapter 或 extra={} 代替 Filter 吗？
可以用 extra={"trace_id": ...}，但你必须每处调用都带上它，跨库调用时无法做到；LoggerAdapter 只对通过该 adapter 发出的日志生效，logging.getLogger("httpx") 打出来的照样没有。Filter + ContextVar 是唯一“零侵入业务代码”的方案。两者也可以叠加：Adapter 用于额外业务字段，Filter 负责 trace_id。
多进程部署（gunicorn -w 4 / uvicorn –workers）会影响吗？
不会，ContextVar 是进程内隔离的，各 worker 各管各的，前提是入口已经按请求生成了新的 trace_id。真正需要额外处理的是跨进程的下游调用：通过 Header 或消息队列的消息属性透传。若用 multiprocessing 派生子进程且需要延续父进程的上下文，显式 set 或把 trace_id 作为启动参数传入，不要指望自动继承。
有没有现成的库可以少写点代码？
常见的有 asgi-correlation-id（Starlette/FastAPI 中间件）和 structlog 的 contextvars 绑定（structlog.contextvars.bind_contextvars），它们内部就是本文这套机制。OpenTelemetry 的 Python SDK 也用 ContextVar 存 span context，若链路追踪已接 OTel，直接读当前 span 的 trace id 注入日志更省事。但无论用哪个库，run_in_executor 不搬运 Context、Handler 级 Filter 这两个坑依然存在，理解了原理才能定位线上问题。