---
title: "Python 异步日志串了请求 ID：用 ContextVar 做请求级隔离"
date: 2026-11-12
slug: python-contextvars-request-logging
canonical: https://kernelranch.com/blog/2026/11/python-contextvars-request-logging
markdown_url: https://kernelranch.com/blog/2026/11/python-contextvars-request-logging.md
tags: ["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()` 碰运气。

```python
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 `contextvars` | https://docs.python.org/3.11/library/contextvars.html | `ContextVar`、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 实测协程交错、过滤器输出、三种线程读取结果和取消后的恢复。
- 示例使用标准库。接入框架时可根据中间件、日志队列和后台任务的实际边界复跑同样的检查。