Python logging 请求 ID 怎样处理

我在给一个同时处理同步任务和异步任务的服务补日志时，最先想做的是在每一行 logger.info() 里手动追加请求 ID。很快就发现这条路会让业务代码到处传参数，漏写一处就无法把一条请求串起来。更麻烦的是，并发请求共享线程时，不能把普通全局变量当作上下文。

更稳妥的做法是：用模块级 ContextVar 保存当前请求的 request_id，在入口设置并在退出时恢复；再用一个 logging.Filter 读取它，把字段写进 LogRecord，由 Formatter 统一输出。官方文档入口：https://docs.python.org/3.15/library/logging.html。

请求 ID 的关键不是“每次打印时拼字符串”，而是让上下文只在当前请求范围内生效，并让日志管道统一补字段。

先把基线和边界说清楚

我会先把问题拆成三个可观察指标：带有 request_id 的日志占比、并发请求之间的串号次数、单条日志增加的额外处理时间。下面的数字是压测记录的示例格式，用来说明应该观察什么，不代表某个固定框架的性能承诺。

这个边界很重要：请求 ID 只用于关联日志，不应放入密码、完整 Cookie 或其他敏感信息；如果日志会跨进程传输，还要在网关或消息协议层明确字段名和生成规则。

先把请求上下文放到 ContextVar

contextvars 提供的是上下文局部状态。它适合保存请求范围的数据，官方文档也说明它可以用于线程和 asyncio 场景。变量放在模块顶层，入口拿到新 ID 后调用 set()，结束时用返回的 token 恢复旧值。

import logging
from contextvars import ContextVar

logger = logging.getLogger(__name__)

# 变量放在模块顶层，避免每次请求重复创建 ContextVar。
request_id_var: ContextVar[str] = ContextVar("request_id", default="-")

def handle_request(request_id: str) -> None:
    # set() 返回 token，退出请求时用它恢复调用前的上下文。
    token = request_id_var.set(request_id)
    try:
        logger.info("开始处理请求")
        do_work()
    finally:
        # 无论业务成功还是抛异常，都不能把本次 ID 留给后续请求。
        request_id_var.reset(token)

def do_work() -> None:
    # 业务函数不用再接收 request_id 参数。
    logger.info("执行业务步骤")

这里的重点不是把所有业务状态都塞进 ContextVar，而是只放少量、生命周期明确的关联字段。嵌套调用时不要直接覆盖后丢弃 token，否则外层请求恢复时会失去原值。

用 Filter 统一补齐日志字段

仅设置 ContextVar 不会自动改变日志文本。要让所有模块的日志都能使用同一个格式，可以在 Handler 上挂一个 Filter，读取当前上下文后给 LogRecord 增加 request_id。这样业务调用仍保持普通的 logger.info() 形式。

class RequestIdFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        # 没有请求上下文时使用占位符，避免 Formatter 因缺字段报错。
        record.request_id = request_id_var.get()
        return True

handler = logging.StreamHandler()
handler.addFilter(RequestIdFilter())
handler.setFormatter(logging.Formatter(
    "%(asctime)s %(levelname)s [request_id=%(request_id)s] %(message)s"
))

root = logging.getLogger()
root.setLevel(logging.INFO)
root.addHandler(handler)

# 这些调用不需要手动拼接 request_id。
logging.getLogger(__name__).info("日志字段由 Filter 统一补齐")

Filter 挂在 Handler 还是 Logger，需要结合输出目标决定。只想让某个输出目的地带请求 ID 时，挂在对应 Handler 上更容易控制；如果多个 Handler 都需要这个字段，就要确保每条记录在格式化前都已经拥有它。logging 官方 Cookbook 还介绍了 LoggerAdapter 和 Filter 两种注入上下文的方式：前者适合显式携带一组固定上下文，后者适合统一读取当前请求状态。

在同步和异步入口设置并恢复

同步入口可以直接使用前面的 try/finally。异步入口同样遵循“设置、调用、恢复”的顺序，不能因为 await 会切换任务，就改用全局变量。ContextVar 的值会随当前上下文进入任务，任务之间仍保持隔离。

import asyncio

async def handle_async(request_id: str) -> None:
    # 每个异步任务设置自己的 ID，避免并发任务互相覆盖。
    token = request_id_var.set(request_id)
    try:
        logger.info("异步请求开始")
        await asyncio.sleep(0)
        logger.info("异步请求完成")
    finally:
        # 任务结束后恢复进入前的上下文，便于复用执行环境。
        request_id_var.reset(token)

async def main() -> None:
    # 两个任务交错执行时，日志仍应分别带 R-17 和 R-18。
    await asyncio.gather(
        handle_async("R-17"),
        handle_async("R-18"),
    )

真正接入框架时，把设置动作放在中间件、请求钩子或任务包装器里即可。线程池、进程池和消息队列边界则要单独确认上下文是否被复制；跨进程时通常应把 request_id 放进消息或任务参数，由消费端重新设置。

用并发场景观察结果和额外开销

我会先用两组交错请求观察日志顺序：同一个 ID 是否贯穿入口、业务函数和异常记录；不同 ID 是否出现互相替换；没有请求上下文的后台任务是否稳定输出 -。这三项比单次顺序调用更能暴露实现问题。

性能上不要把 Filter 说成“零成本”。每条记录都会经过过滤器和格式化，真实代价取决于日志级别、Handler 数量、输出目的地以及消息量。可以分别测量关闭 INFO、开启 INFO、并发 任务和纯内存 Handler 四种情况，把日志处理时间与业务耗时分开记录；若高频路径对延迟敏感，先降低日志级别或采样，再考虑更复杂的异步日志方案。

• 出现串号：检查是否存在全局可变变量、是否忘记保存/恢复 token。
• 出现格式化异常：检查所有输出 Handler 是否都能拿到 request_id。
• 后台任务显示 -：确认任务是否在请求上下文外创建，以及是否需要显式传递关联 ID。

结论是：Python logging 请求 ID 最小可靠方案是 ContextVar + Filter + Formatter。ContextVar 负责隔离生命周期，Filter 负责统一注入，Formatter 负责稳定输出；三者职责分开后，既不用污染业务函数签名，也能让并发排查有清晰的关联线索。

相关问题

为什么不直接把 request_id 作为每次日志调用的参数？

小脚本可以这样做，但在多层调用和第三方库日志中很容易漏传。统一注入能减少调用点改动，不过入口仍要负责生成、设置和恢复请求上下文。

LoggerAdapter 和 Filter 应该选哪个？

上下文是固定且由调用方明确持有时，LoggerAdapter 更直观；上下文来自当前请求、希望所有模块共享同一字段时，Filter 更适合。

线程池或进程池里还能直接读取这个 ID 吗？

不要默认假设跨执行边界会自动复制。把 ID 作为任务数据传入，在目标线程或进程入口重新设置，并在结束时恢复。

事实来源：https://docs.python.org/3.15/library/logging.html https://docs.python.org/3.15/howto/logging-cookbook.html https://docs.python.org/3.15/library/contextvars.html