KernelRanch
返回文章列表

Python 异步日志串了请求 ID：用 ContextVar 做请求级隔离

2026-11-12 #python#asyncio#logging#可观测性

请求 A 等数据库时让出执行权，请求 B 把全局 request_id 改成自己的。A 恢复后打出的日志就会带上 B 的 ID。异步服务可以在请求入口用 ContextVar 绑定 ID，通过日志过滤器自动填充，并在 finally 用 token 恢复旧值。这样业务函数只需正常打日志。

1. 让两个协程按固定顺序交错

保存为 request_log_lab.py，运行 python request_log_lab.py。实验用 asyncio.Event 规定顺序，A 必须等 B 绑定之后才能写日志，无须靠 sleep() 碰运气。

import asyncio
import logging
from contextlib import contextmanager
from contextvars import ContextVar, copy_context


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


@contextmanager
def global_scope(value):
    global global_id
    previous, global_id = global_id, value
    try:
        yield
    finally:
        global_id = previous


@contextmanager
def request_scope(value):
    token = request_id.set(value)
    try:
        yield
    finally:
        request_id.reset(token)


class RequestIdFilter(logging.Filter):
    def filter(self, record):
        record.request_id = request_id.get()
        return True


async def interleave(scope, read_id):
    a_ready, b_ready, a_read, b_done = [asyncio.Event() for _ in range(4)]
    seen = []

    async def handle_a():
        with scope("req-a"):
            a_ready.set()
            await b_ready.wait()
            seen.append(("A", read_id()))
            a_read.set()
            await b_done.wait()

    async def handle_b():
        await a_ready.wait()
        with scope("req-b"):
            b_ready.set()
            await a_read.wait()
            seen.append(("B", read_id()))
        b_done.set()

    await asyncio.wait_for(asyncio.gather(handle_a(), handle_b()), timeout=5)
    return seen


async def main():
    wrong = await interleave(global_scope, lambda: global_id)
    right = await interleave(request_scope, request_id.get)
    assert wrong == [("A", "req-b"), ("B", "req-b")]
    assert right == [("A", "req-a"), ("B", "req-b")]
    print("global:", wrong)
    print("context:", right)

    logger = logging.getLogger("request_lab")
    logger.setLevel(logging.INFO)
    logger.propagate = False
    handler = logging.StreamHandler()
    handler.addFilter(RequestIdFilter())
    handler.setFormatter(logging.Formatter("request=%(request_id)s %(message)s"))
    logger.addHandler(handler)

    with request_scope("req-a"):
        logger.info("handling request")
        via_to_thread = await asyncio.to_thread(request_id.get)
        loop = asyncio.get_running_loop()
        via_executor = await loop.run_in_executor(None, request_id.get)
        ctx = copy_context()
        via_copied_context = await loop.run_in_executor(None, ctx.run, request_id.get)
        assert (via_to_thread, via_executor, via_copied_context) == ("req-a", "-", "req-a")
    logger.info("outside request")
    print("threads:", via_to_thread, via_executor, via_copied_context)

    entered = asyncio.Event()

    async def cancelled_request():
        try:
            with request_scope("req-cancelled"):
                entered.set()
                await asyncio.Future()
        finally:
            # 在被取消的同一个任务里验证清理。
            assert request_id.get() == "-"

    task = asyncio.create_task(cancelled_request())
    await asyncio.wait_for(entered.wait(), timeout=5)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("cancellation: context restored")
    else:
        raise AssertionError("expected task cancellation")


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

本地实测，A 用全局变量读到 req-b，用 ContextVar 则读到 req-a。过滤器在处理请求时写出 request=req-a handling request，退出请求作用域后写出 request=- outside request。取消测试在被取消的任务内部断言旧值已经恢复。

2. 线程和后台任务各看一次传播行为

调用方式本例结果接入方式
await asyncio.to_thread(fn)线程读到 req-a适合在阻塞 I/O 中继续使用当前请求 ID
loop.run_in_executor(None, fn)默认线程池里读到 -需要传播时，为本次提交创建一个 copy_context()，再以 ctx.run 调用
asyncio.create_task(coro)默认复制创建任务时的当前上下文请求内子任务可沿用 ID。独立后台任务可在 Python 3.11 中传 context=Context()，并自行绑定任务 ID

请求 ID 适合用字符串保存。复制上下文时，变量引用的可变对象仍可能共享。把字典放进 ContextVar 后在多个子任务原地修改，仍会产生共享状态问题。线程池每次提交各自复制上下文，避免多个线程同时进入同一个 Context。

3. 把绑定放在入口，把注入放在发日志的一侧

在请求中间件里先确定 ID，再用 with request_scope(id) 包住后续处理，包括异常路径。接收外部 ID 时，限制允许的字符和长度，或直接使用服务端生成的 ID。日志中的请求 ID 只用于关联记录。

把 RequestIdFilter 挂到需要该字段的 handler。若日志经过 QueueHandler 发往监听线程，在生产日志的一侧先填 LogRecord。监听线程读取的是自己的上下文，只需格式化已经填好的字段。

小结

请求入口绑定 ID，日志发出时注入 ID，退出时用 token 恢复。验收至少覆盖两个交错协程、一次取消，以及服务中实际用到的线程池调用方式。

数据来源与检索时间

检索时间 2026-11-12。

来源URL对应内容
Python 3.11 contextvarshttps://docs.python.org/3.11/library/contextvars.htmlContextVar、token、copy_context() 和 asyncio 集成
Python 3.11 asyncio 任务https://docs.python.org/3.11/library/asyncio-task.html任务上下文、取消和 to_thread()
Python 3.11 asyncio 事件循环https://docs.python.org/3.11/library/asyncio-eventloop.html#asyncio.loop.run_in_executor向执行器提交工作
Python 日志手册https://docs.python.org/3.11/howto/logging-cookbook.html#use-of-contextvars用上下文变量填充日志字段

核验说明

  • 在 Python 3.11.9 实测协程交错、过滤器输出、三种线程读取结果和取消后的恢复。
  • 示例使用标准库。接入框架时可根据中间件、日志队列和后台任务的实际边界复跑同样的检查。
#python#asyncio#logging#可观测性
返回文章列表