ai

1. 什么是 ContextVar？ #

ContextVar 是 Python 3.7+ 标准库 contextvars 提供的上下文变量。
你可以把它理解成「带隔离能力的全局变量」：写法像全局变量一样随处可读写，但每个执行环境各自一份值，互不干扰。
最常见用途：在异步 Web、异步任务里存放「当前请求的用户 ID、请求 ID、语言」等，而不必一层层当函数参数往下传。
模块名是 contextvars（复数），常用类型是 ContextVar（单数）。

一句话记忆：全局变量全程序共享一份；ContextVar 按「上下文」各存一份。

2. 前置知识：先弄清三个概念 #

2.1 什么是「全局变量」？ #

写在函数外面的变量，整个模块里都能直接用。
优点：调用链再深也能直接读到。
缺点：所有调用方共用同一份数据。多任务同时跑时，A 改了值，B 也会读到被改后的值，容易「串台」。

# 说明：定义一个模块级全局变量，初始用户名为 guest
current_user = "guest"

# 说明：定义函数，打印当前全局变量里的用户名
def show_user():
    # 说明：直接读取全局变量（无需通过参数传入）
    print(f"当前用户: {current_user}")

# 说明：修改全局变量
current_user = "alice"

# 说明：调用函数，会打印 alice
show_user()

2.2 什么是「线程」和 threading.local？ #

线程：操作系统里可以并行执行的一条执行路径。多线程程序里，多条路径可能同时跑。
threading.local()：给每个线程单独存一份数据。线程 A 存的值，线程 B 读不到。
局限：asyncio 里很多协程往往跑在同一个线程里，此时 threading.local 分不开它们，数据仍会互相覆盖。

# 说明：导入线程相关模块
import threading

# 说明：创建一个线程局部存储对象
local = threading.local()

# 说明：定义在子线程中运行的函数
def worker(name):
    # 说明：给当前线程自己的存储写入名字（不影响其他线程）
    local.name = name
    # 说明：打印当前线程读到的名字
    print(f"线程读到: {local.name}")

# 说明：创建两个线程，分别写入 alice 和 bob
t1 = threading.Thread(target=worker, args=("alice",))
t2 = threading.Thread(target=worker, args=("bob",))

# 说明：启动两个线程
t1.start()
t2.start()

# 说明：等待两个线程都执行结束
t1.join()
t2.join()

2.3 什么是「协程」？ #

协程：用 async def 定义的可暂停函数。多个协程可在同一线程里交替执行（遇到 await 就让出执行权）。
asyncio 创建新任务时，会把当前上下文复制一份给新任务。之后两边各自改自己的背包，互不影响。

# 说明：导入 asyncio，用于运行异步协程
import asyncio

# 说明：定义一个简单的异步函数
async def hello(name):
    # 说明：打印问候语
    print(f"你好, {name}")
    # 说明：异步休眠 0.1 秒（让出执行权，不卡住整个程序）
    await asyncio.sleep(0.1)
    # 说明：休眠结束后再打印一句
    print(f"{name} 说再见")

# 说明：定义主协程，并发跑两个子任务
async def main():
    # 说明：同时等待两个协程完成
    await asyncio.gather(hello("小明"), hello("小红"))

# 说明：启动事件循环并运行 main
asyncio.run(main())

3. 为什么需要 ContextVar？ #

先用「全局变量串台」说明问题，再用 ContextVar 给出正确写法。

3.1 问题：全局变量在并发里会串台 #

两个异步任务都改同一个全局变量时，后改的会覆盖先改的，打印结果混乱。

# 说明：导入 asyncio
import asyncio

# 说明：用普通全局变量记录「当前用户」
current_user = "guest"

# 说明：定义处理请求的异步函数
async def handle_request(user):
    # 说明：声明要修改模块级全局变量
    global current_user
    # 说明：把全局变量改成当前请求的用户
    current_user = user
    # 说明：模拟耗时 I/O，期间其他协程也可能改全局变量
    await asyncio.sleep(0.1)
    # 说明：再次读取全局变量（此时可能已被别的协程改掉）
    print(f"期望用户={user}, 实际读到={current_user}")

# 说明：定义主协程，并发处理两个请求
async def main():
    # 说明：同时处理 alice 和 bob 两个请求
    await asyncio.gather(
        handle_request("alice"),
        handle_request("bob"),
    )

# 说明：运行主协程
asyncio.run(main())

可能输出（顺序不固定，但「串台」很常见）：

期望用户=alice, 实际读到=bob
期望用户=bob, 实际读到=bob

3.2 解决：用 ContextVar 隔离每个任务的数据 #

每个任务有自己的上下文副本，set / get 只影响当前上下文。

# 说明：导入 asyncio
import asyncio
# 说明：导入 contextvars 模块
import contextvars

# 说明：创建一个名为 user 的上下文变量，默认值为 guest
user_var = contextvars.ContextVar("user", default="guest")

# 说明：定义处理请求的异步函数
async def handle_request(user):
    # 说明：在当前任务的上下文中设置用户名
    user_var.set(user)
    # 说明：模拟耗时 I/O
    await asyncio.sleep(0.1)
    # 说明：从当前上下文读取用户名（不会被其他任务覆盖）
    print(f"期望用户={user}, 实际读到={user_var.get()}")

# 说明：定义主协程
async def main():
    # 说明：并发处理两个请求
    await asyncio.gather(
        handle_request("alice"),
        handle_request("bob"),
    )

# 说明：运行主协程
asyncio.run(main())

输出示例：

期望用户=alice, 实际读到=alice
期望用户=bob, 实际读到=bob

4. 创建与读写 ContextVar #

先用 ContextVar(名字, default=默认值) 创建变量对象（通常放在模块顶部，全程序共用这一个对象）。
set(值)：写入当前上下文。
get()：读取当前上下文中的值；若从未设置且没有默认值，会抛出 LookupError。
建议总是提供 default

# 说明：导入 contextvars 模块
import contextvars

# 说明：创建上下文变量 request_id，默认值为 unknown
request_id_var = contextvars.ContextVar("request_id", default="unknown")

# 说明：尚未 set 时，get 会返回默认值
print(request_id_var.get())

# 说明：在当前上下文中写入请求 ID
request_id_var.set("req-1001")

# 说明：再次读取，得到刚写入的值
print(request_id_var.get())

运行结果：

unknown
req-1001

5. 跨函数读取：不必层层传参 #

ContextVar 的一大好处：中间函数不必声明「用户 ID」参数，深层函数也能直接 get()。
适合「整条调用链都可能需要、但又不想污染每个函数签名」的数据。

# 说明：导入 contextvars 模块
import contextvars

# 说明：创建用户名字上下文变量，默认匿名
username_var = contextvars.ContextVar("username", default="匿名")

# 说明：最深层的工具函数：直接从上下文取用户名
def write_log(message):
    # 说明：读取当前上下文中的用户名
    user = username_var.get()
    # 说明：打印带用户名的日志
    print(f"[{user}] {message}")

# 说明：中间层业务函数：也不接收用户名参数
def process_order(order_id):
    # 说明：调用日志函数（用户名从上下文来）
    write_log(f"开始处理订单 {order_id}")
    # 说明：模拟业务处理
    write_log(f"订单 {order_id} 处理完成")

# 说明：入口：先 set，再调用深层函数
def main():
    # 说明：把当前用户写入上下文
    username_var.set("小明")
    # 说明：处理订单，中间无需传递用户名
    process_order("A001")

# 说明：运行入口函数
main()

运行结果：

[小明] 开始处理订单 A001
[小明] 订单 A001 处理完成

6. Token 与 reset：改完再恢复 #

set() 会返回一个 Token（令牌），记录「设置之前的旧状态」。
reset(token) 可以把变量恢复到这次 set 之前的样子。
适合「临时改一下，离开前必须还原」的场景，例如函数内部临时切换语言再恢复。

# 说明：导入 contextvars 模块
import contextvars

# 说明：创建语言上下文变量，默认中文
lang_var = contextvars.ContextVar("lang", default="zh")

# 说明：定义一个临时切换语言的函数
def with_english():
    # 说明：设置英文，并保存返回的 token，方便之后还原
    token = lang_var.set("en")
    # 说明：打印当前语言
    print(f"函数内语言: {lang_var.get()}")
    # 说明：用 token 恢复到 set 之前的值
    lang_var.reset(token)
    # 说明：打印恢复后的语言
    print(f"恢复后语言: {lang_var.get()}")

# 说明：先把当前语言设为中文
lang_var.set("zh")

# 说明：打印进入函数前的语言
print(f"进入前语言: {lang_var.get()}")

# 说明：调用临时切换函数
with_english()

# 说明：打印离开函数后的语言（仍是 zh）
print(f"离开后语言: {lang_var.get()}")

运行结果：

进入前语言: zh
函数内语言: en
恢复后语言: zh
离开后语言: zh

7. 在 asyncio 里如何隔离？ #

这是 ContextVar 最重要的场景：同一线程里多个协程任务，各自有一份值。

7.1 任务创建时会复制上下文 #

使用 asyncio.create_task(...) 时，会立刻复制当时的上下文给新任务。
之后你在主协程里再 set 新值，不会改到已经创建好的任务。

# 说明：导入 asyncio
import asyncio
# 说明：导入 contextvars
import contextvars

# 说明：创建用户上下文变量
user_var = contextvars.ContextVar("user", default="guest")

# 说明：定义子任务：打印自己上下文里的用户
async def worker(tag):
    # 说明：读取当前任务上下文中的用户
    print(f"{tag} 开始: {user_var.get()}")
    # 说明：异步休眠，期间主协程可能改自己的上下文
    await asyncio.sleep(0.05)
    # 说明：再次读取，仍应是创建任务时复制来的值
    print(f"{tag} 结束: {user_var.get()}")

# 说明：定义主协程
async def main():
    # 说明：先把主上下文设为 alice
    user_var.set("alice")
    # 说明：创建任务 A（此时复制到的是 alice）
    task_a = asyncio.create_task(worker("任务A"))
    # 说明：创建任务 B（此时复制到的仍是 alice）
    task_b = asyncio.create_task(worker("任务B"))
    # 说明：主协程把自己的值改成 bob（不影响已创建的任务）
    user_var.set("bob")
    # 说明：打印主协程自己的值
    print(f"主协程: {user_var.get()}")
    # 说明：等待两个子任务结束
    await asyncio.gather(task_a, task_b)

# 说明：运行主协程
asyncio.run(main())

运行结果示例：

主协程: bob
任务A 开始: alice
任务B 开始: alice
任务A 结束: alice
任务B 结束: alice

7.2 让每个任务一创建就有自己的值 #

常见写法：在子协程一开头就 set，这样每个任务写入的是自己的上下文。

# 说明：导入 asyncio
import asyncio
# 说明：导入 contextvars
import contextvars

# 说明：创建请求 ID 上下文变量
req_var = contextvars.ContextVar("req_id", default="-")

# 说明：业务函数：从上下文读请求 ID 并打印
def log(msg):
    # 说明：读取当前上下文中的请求 ID
    print(f"[{req_var.get()}] {msg}")

# 说明：模拟一次请求处理
async def handle(req_id):
    # 说明：进入本任务后立刻写入自己的请求 ID
    req_var.set(req_id)
    # 说明：记录开始日志
    log("开始处理")
    # 说明：模拟 IO
    await asyncio.sleep(0.05)
    # 说明：记录结束日志
    log("处理结束")

# 说明：主协程并发两个请求
async def main():
    # 说明：并发运行两个带不同请求 ID 的任务
    await asyncio.gather(
        handle("R-1"),
        handle("R-2"),
    )

# 说明：启动
asyncio.run(main())

运行结果示例（两行的交错顺序可能不同）：

[R-1] 开始处理
[R-2] 开始处理
[R-1] 处理结束
[R-2] 处理结束

8. copy_context 与 run：手动开一个「隔离沙箱」 #

copy_context()：复制当前整个上下文，得到一个 Context 对象。
ctx.run(函数)：在这个副本里执行函数；函数内的 set 只改副本，外面看不到。
日常异步代码里，create_task 已经自动复制；

# 说明：导入 contextvars
import contextvars

# 说明：创建上下文变量，默认 outer
name_var = contextvars.ContextVar("name", default="outer")

# 说明：先在外层上下文写入 outer-value
name_var.set("outer-value")

# 说明：复制当前上下文，得到一份独立副本
ctx = contextvars.copy_context()

# 说明：定义将在副本上下文中运行的函数
def inside():
    # 说明：副本刚开始时，值和复制那一刻一致
    print(f"沙箱内读取: {name_var.get()}")
    # 说明：在沙箱内修改（只影响副本）
    name_var.set("inner-value")
    # 说明：确认沙箱内已变成新值
    print(f"沙箱内修改后: {name_var.get()}")

# 说明：在复制出的上下文中运行 inside
ctx.run(inside)

# 说明：外层上下文仍然是原来的值
print(f"外层读取: {name_var.get()}")

运行结果：

沙箱内读取: outer-value
沙箱内修改后: inner-value
外层读取: outer-value

9. 与 threading.local 怎么选？ #

对比项	`threading.local`	`ContextVar`
隔离单位	线程	上下文（含协程任务）
同线程多协程	会互相覆盖	可各自独立
典型场景	纯多线程、无 asyncio	asyncio / FastAPI 等异步程序
建议	不做异步时可用	写异步就优先用它

只开多线程、不用 asyncio：threading.local 够用。
用了 asyncio（或两者混用）：优先 ContextVar，否则协程之间容易串数据。

10. 易踩的坑 #

10.1 可变对象：改内容不等于换值 #

ContextVar 隔离的是「变量指向哪个对象」。
若存的是 list / dict，多个上下文可能仍指向同一个对象；对列表 append 会互相看见。
建议：优先存 str、int、tuple 等不可变数据；需要字典时，改用新字典整体 set 进去。

# 说明：导入 contextvars
import contextvars

# 说明：创建用于存放标签列表的上下文变量，默认空元组
tags_var = contextvars.ContextVar("tags", default=())

# 说明：错误示范：把同一个 list 对象 set 进去再原地修改
bad_list = ["a"]
# 说明：写入可变列表
tags_var.set(bad_list)
# 说明：原地修改列表内容（所有引用该列表的地方都会看到变化）
bad_list.append("b")
# 说明：打印，会看到 a 和 b
print("可变对象陷阱:", tags_var.get())

# 说明：推荐写法：每次用新的不可变对象整体替换
tags_var.set(("a",))
# 说明：需要增加元素时，创建新元组再 set
tags_var.set(tags_var.get() + ("b",))
# 说明：打印推荐写法的结果
print("推荐写法:", tags_var.get())

运行结果：

可变对象陷阱: ['a', 'b']
推荐写法: ('a', 'b')

10.2 忘记设默认值 #

没有 default 且从未 set 时，get() 会抛 LookupError。
也可以 get(默认值) 临时指定，但创建时写好 default 更稳妥。

# 说明：导入 contextvars
import contextvars

# 说明：创建一个没有默认值的上下文变量
no_default = contextvars.ContextVar("no_default")

# 说明：尝试安全读取：先捕获 LookupError
try:
    # 说明：从未 set 且无 default，这里会报错
    print(no_default.get())
# 说明：捕获查找失败异常
except LookupError:
    # 说明：打印提示信息
    print("尚未设置，且没有默认值")

# 说明：调用 get 时传入临时默认值，避免异常
print(no_default.get("临时默认"))

运行结果：

尚未设置，且没有默认值
临时默认

11. 综合案例：异步请求日志 #

把前面的知识串起来：入口写入请求 ID，深层函数只负责打日志，多个请求并发也不会串。

# 说明：导入 asyncio
import asyncio
# 说明：导入 contextvars
import contextvars

# 说明：请求 ID 上下文变量
request_id_var = contextvars.ContextVar("request_id", default="-")
# 说明：用户名上下文变量
username_var = contextvars.ContextVar("username", default="guest")

# 说明：通用日志函数：自动带上请求 ID 和用户名
def log(message):
    # 说明：从上下文读取请求 ID
    rid = request_id_var.get()
    # 说明：从上下文读取用户名
    user = username_var.get()
    # 说明：打印统一格式的日志
    print(f"{rid} | {user} | {message}")

# 说明：模拟查询数据库（深层函数，不接收用户参数）
async def query_db():
    # 说明：记录查询开始
    log("查询数据库...")
    # 说明：模拟数据库耗时
    await asyncio.sleep(0.05)
    # 说明：记录查询结束
    log("查询完成")

# 说明：模拟一次完整请求
async def handle_request(request_id, username):
    # 说明：进入请求后写入本请求的上下文信息
    request_id_var.set(request_id)
    # 说明：写入当前用户名
    username_var.set(username)
    # 说明：记录请求开始
    log("收到请求")
    # 说明：调用深层业务（无需传参）
    await query_db()
    # 说明：记录请求结束
    log("返回响应")

# 说明：主程序：并发两个请求
async def main():
    # 说明：同时处理两个用户的请求
    await asyncio.gather(
        handle_request("REQ-001", "alice"),
        handle_request("REQ-002", "bob"),
    )

# 说明：作为脚本直接运行时启动
if __name__ == "__main__":
    # 说明：运行主协程
    asyncio.run(main())

运行结果示例（同一请求的三行日志中，请求 ID 与用户始终一致）：

REQ-001 | alice | 收到请求
REQ-002 | bob | 收到请求
REQ-001 | alice | 查询数据库...
REQ-002 | bob | 查询数据库...
REQ-001 | alice | 查询完成
REQ-002 | bob | 查询完成
REQ-001 | alice | 返回响应
REQ-002 | bob | 返回响应

12. 小结 #

ContextVar = 按上下文隔离的「类全局变量」，特别适合 asyncio。
常用三板斧：ContextVar(..., default=...) → set(值) → get()。
需要还原时：保存 token = var.set(...)，结束用 var.reset(token)。
create_task 会复制上下文；任务内再 set，只影响自己。
少存可变对象；优先存字符串、数字等简单值。
同步多线程且不用异步时，继续用 threading.local 即可；一上异步，优先换 ContextVar。 )

导航菜单

1. 什么是 ContextVar？ #

2. 前置知识：先弄清三个概念 #

2.1 什么是「全局变量」？ #

2.2 什么是「线程」和 threading.local？ #

2.3 什么是「协程」？ #

3. 为什么需要 ContextVar？ #

3.1 问题：全局变量在并发里会串台 #

3.2 解决：用 ContextVar 隔离每个任务的数据 #

4. 创建与读写 ContextVar #

5. 跨函数读取：不必层层传参 #

6. Token 与 reset：改完再恢复 #

7. 在 asyncio 里如何隔离？ #

7.1 任务创建时会复制上下文 #

7.2 让每个任务一创建就有自己的值 #

8. copy_context 与 run：手动开一个「隔离沙箱」 #

9. 与 threading.local 怎么选？ #

10. 易踩的坑 #

10.1 可变对象：改内容不等于换值 #

10.2 忘记设默认值 #

11. 综合案例：异步请求日志 #

12. 小结 #