适用场景
FastAPI、Django ASGI 或自研 asyncio 服务会在同一线程交错处理多个请求。团队希望所有日志自动带上 request_id，却发现并发压测时标识偶尔串到另一个请求，或请求结束后启动的后台任务仍携带旧用户上下文。
本文用 Python 标准库复现并修复这个问题。重点不是某个 Web 框架的中间件写法，而是建立一条可测试的上下文生命周期：入口绑定、业务读取、子任务继承、线程边界传播，以及出口清理。
现象描述
最常见的错误实现是把当前请求标识放进模块全局变量：
current_request_id = "-"
async def handle_request(request_id: str) -> None:
global current_request_id
current_request_id = request_id
await call_dependency()
logger.info("请求完成", extra={"request_id": current_request_id})
协程在 await 处让出执行权后，另一个请求会覆盖全局变量。线程本地变量也不能解决同一线程内多个协程交错执行的问题。典型证据包括：访问日志中的请求 ID 正确，但业务层日志出现另一请求的 ID；低并发无法复现，提高并发并增加 I/O 等待后复现率上升。
排查时先确认串号发生在哪个边界：
- 请求入口是否生成或校验了标识，长度与字符集是否受限？
- 标识是否保存在全局变量、单例对象字段或可变字典中？
- 是否在
finally中恢复上下文，而不是简单写回默认值？ - 后台任务是在请求上下文存在时创建，还是在清理后创建？
- 同步函数通过
asyncio.to_thread()、run_in_executor()还是自建线程池执行？
不要把 Authorization、Cookie、手机号或完整请求体当作关联标识。日志只应保存随机、短期、无业务含义的 ID。
原因：线程隔离不等于协程隔离
contextvars.ContextVar 为每个执行上下文保存独立值，适合在异步任务中传递请求级状态。asyncio.Task 创建时会复制当前上下文，因此同一请求创建的普通子任务能继承当时的 request_id，之后其他请求修改自己的上下文不会污染它。
但“自动继承”也有边界：在请求上下文中创建的长期后台任务会把该上下文快照带到请求结束之后；asyncio.to_thread() 会传播当前上下文，而低层的执行器接口不能被当作等价替代。上下文解决的是传播与隔离，不负责决定哪些任务应该继承敏感状态。
实现：绑定、读取和恢复上下文
将以下内容保存为 request_context.py：
"""提供请求标识的绑定、日志注入与后台任务隔离。"""
import asyncio
import contextvars
import logging
import re
import uuid
from collections.abc import Awaitable, Callable
from typing import TypeVar
REQUEST_ID_PATTERN = re.compile(r"^[A-Za-z0-9._-]{1,64}$")
request_id_var: contextvars.ContextVar[str] = contextvars.ContextVar(
"request_id",
default="-",
)
T = TypeVar("T")
def normalize_request_id(candidate: str | None) -> str:
"""接受安全的上游标识，否则生成新的随机标识。"""
if candidate is not None and REQUEST_ID_PATTERN.fullmatch(candidate):
return candidate
return uuid.uuid4().hex
class RequestContextFilter(logging.Filter):
"""把当前请求标识写入每条日志记录。"""
def filter(self, record: logging.LogRecord) -> bool:
record.request_id = request_id_var.get()
return True
async def run_in_request(
candidate: str | None,
operation: Callable[[], Awaitable[T]],
) -> T:
"""在一次请求的生命周期内绑定标识，并在出口恢复旧上下文。"""
request_id = normalize_request_id(candidate)
token = request_id_var.set(request_id)
try:
return await operation()
finally:
request_id_var.reset(token)
def create_background_task(
coroutine: Awaitable[object],
*,
name: str,
) -> asyncio.Task[object]:
"""创建不继承请求上下文的后台任务。"""
clean_context = contextvars.copy_context()
clean_context.run(request_id_var.set, "-")
return asyncio.create_task(coroutine, name=name, context=clean_context)
set() 返回的 token 记录了设置前的值。必须用 reset(token) 恢复，因为中间件可能嵌套：测试工具、反向代理适配层和业务入口都可能临时绑定上下文。直接在出口执行 set("-") 会破坏外层上下文。
入口标识属于外部输入。示例只接受 1 到 64 位字母、数字、点、下划线和连字符，换行符等日志注入字符会被拒绝并替换为随机值。生产系统还应由可信网关决定是否接受客户端传入的请求 ID。
日志格式可以这样配置：
handler = logging.StreamHandler()
handler.addFilter(RequestContextFilter())
handler.setFormatter(
logging.Formatter(
"%(asctime)s %(levelname)s request_id=%(request_id)s %(message)s"
)
)
logger = logging.getLogger("service")
logger.setLevel(logging.INFO)
logger.addHandler(handler)
过滤器应挂在真正输出日志的 handler 上。若只挂在某个子 logger，而第三方 logger 的记录由根 handler 输出，格式化器仍可能因为缺少 request_id 字段报错。
Web 中间件如何接入
框架中间件只负责把请求生命周期包装进 run_in_request()。以下伪代码展示边界，具体请求头读取和响应头写入应使用所选框架的 API：
async def request_middleware(request, call_next):
candidate = request.headers.get("X-Request-ID")
async def dispatch():
response = await call_next(request)
response.headers["X-Request-ID"] = request_id_var.get()
return response
return await run_in_request(candidate, dispatch)
响应头必须在上下文恢复前读取。错误响应也要经过同一出口，否则最需要关联日志的 500 请求反而缺少标识。若服务还要调用下游 HTTP 接口，应在发出请求时读取 request_id_var.get() 并写入约定的追踪头，但不要无条件信任来自公网的同名头。
验证：并发隔离、异常清理与后台任务
将以下内容保存为 test_request_context.py：
"""验证请求上下文在并发、异常和后台任务中的生命周期。"""
import asyncio
import unittest
from request_context import (
create_background_task,
request_id_var,
run_in_request,
)
class RequestContextTests(unittest.IsolatedAsyncioTestCase):
"""为每个用例使用独立事件循环。"""
async def test_concurrent_requests_are_isolated(self) -> None:
"""两个交错请求始终读取各自的标识。"""
barrier = asyncio.Barrier(2)
async def read_twice() -> tuple[str, str]:
before = request_id_var.get()
await barrier.wait()
await asyncio.sleep(0)
return before, request_id_var.get()
first, second = await asyncio.gather(
run_in_request("req-a", read_twice),
run_in_request("req-b", read_twice),
)
self.assertEqual(first, ("req-a", "req-a"))
self.assertEqual(second, ("req-b", "req-b"))
async def test_exception_restores_outer_context(self) -> None:
"""业务异常传播后恢复进入前的上下文。"""
outer_token = request_id_var.set("outer")
async def fail() -> None:
raise RuntimeError("模拟业务失败")
try:
with self.assertRaises(RuntimeError):
await run_in_request("inner", fail)
self.assertEqual(request_id_var.get(), "outer")
finally:
request_id_var.reset(outer_token)
async def test_background_task_has_clean_context(self) -> None:
"""长期后台任务不携带已结束请求的标识。"""
observed: list[str] = []
async def start_background() -> None:
async def worker() -> None:
observed.append(request_id_var.get())
task = create_background_task(worker(), name="refresh_cache")
await task
await run_in_request("req-private", start_background)
self.assertEqual(observed, ["-"])
if __name__ == "__main__":
unittest.main()
执行：
python -m unittest -v test_request_context
三项测试分别覆盖并发交错、异常出口和长期任务隔离。不要只验证顺序执行的两个请求；那种测试无法暴露全局变量被覆盖的问题。
线程池与后台任务的注意事项
需要把同步函数移到线程执行时，优先使用 await asyncio.to_thread(func, *args)，并通过测试确认目标 Python 版本会传播当前上下文。如果使用框架自己的线程池或任务队列，不要推测其行为，应把 request_id 作为显式参数传入边界，或按照该组件文档复制上下文。
对于 Celery、消息队列等进程外任务，ContextVar 不会跨进程传播。应在消息元数据中写入新的 job_id 或经过审核的追踪标识，由消费者重新绑定并在处理结束后清理。请求 ID 不应被当作任务幂等键，因为重试与一次 HTTP 请求并不是同一个生命周期。
长期后台任务还必须被任务管理器持有引用，并在服务关闭时等待或取消。本文的 create_background_task() 只演示上下文隔离，不替代任务持久化、失败重试和优雅关闭机制。
上线与观测
先在一个低风险接口灰度，记录固定中文消息与稳定结构化字段，例如 request_id、route、status_code、duration_ms。不要把动态值拼进消息，也不要记录原始请求头集合。
上线后可用以下信号验收：同一请求链路的标识是否一致；并发请求之间是否出现相同但不应相同的标识；无请求的定时任务是否保持默认值；异常请求结束后下一条启动日志是否仍残留旧标识。若系统已有 OpenTelemetry trace，应优先复用标准 trace 上下文，避免再维护一套相互矛盾的关联体系。
总结
异步日志串号通常不是日志库本身的问题，而是请求状态被放进了错误的共享位置。使用 ContextVar 后仍要明确四个责任：入口校验并绑定、业务只读当前值、出口用 token 恢复、长期任务主动隔离。再用真实交错的并发测试和异常测试验证，才能保证关联标识既不串号，也不会在请求结束后泄漏到不相关工作中。
Discussion
评论