如果你已经写过一段时间的 Python 并发代码大概率遇到过下面这类诡异现象同一个函数单独跑没有任何问题一放进asyncio.gather()里全局变量的值就开始变得不可预测或者信心满满地写下一个global counter却收到一条UnboundLocalError又或者闭包里那个循环变量i等到任务真正执行时所有协程读到的都是同一个数。问题往往不在业务逻辑本身而在于命名空间的边界没画好再加上协程的调度时机刚好把边界问题放大了。命名空间和协程一个管名字如何解析一个管执行流程如何切换看起来互不相关。可并发代码写多了就会明白两者共同决定了程序里每个状态变量“在哪里可见、在哪一刻可见”。命名空间划定数据可见的范围协程决定数据被访问的时机。把这套组合理顺高并发系统才真正有了可靠的地基。这篇不是教科书式的入门介绍而是从实战排障视角把两个主题拆开、揉碎再拼回去最后给出一套可直接参考的并发骨架。1. 为什么命名空间与协程必须放在一起讲1.1 命名空间决定了状态“在哪里被看见”命名空间本质上就是一张“名字到对象”的映射表。你写x 1是在当前作用域的命名空间里注册一个名字你读x是顺着作用域链逐层查找直到找到为止。这听起来很简单但一旦程序里有多个并发执行单元问题立刻变得微妙同一个名字在协程 A 的局部命名空间里指向一份数据在协程 B 里可能指向另一份可如果这个名字被定义在模块级所有协程看到的都是同一份对象。我常用一个比喻来理解这个事。协程相当于同一家公司里并行办公的多个员工而模块级全局变量就是公司前台那块公共白板。每个员工的任务单是局部命名空间员工抬头看白板是全局命名空间。如果两个员工同时往白板上同一个数字后面追加内容就会出现“一个擦了另一个的笔迹”的情况。更麻烦的是白板上的内容谁都能写你很难追查是哪一笔出了问题。并发系统里的大部分状态污染根因都是数据被放进了过宽的命名空间。这里有个反直觉的点Python 是单线程语言很多人因此觉得“单线程不存在数据竞争”。实际上单线程只是消除了同一时刻并行执行并没有消除交错执行。只要程序被切分成多个任务任务之间的状态共享就可能产生竞争只是竞争的形式从“同时读写”变成了“在某个检查点之间读写”。这个检查点就是协程里的await。1.2 await 是检查点状态“在什么时刻被看见”await表面上只是等待一个结果真正的含义是挂起当前协程把控制权还给事件循环。事件循环拿到控制权后可以调度另一个协程继续执行。于是一条看似连续的代码会被await切成几段tmp shared_counter await asyncio.sleep(0) # 让出控制权 shared_counter tmp 1第一行读shared_counter第三行写shared_counter中间隔了一个await。在这段时间里另一个协程可能已经把shared_counter改了。你以为自己在做一次原子操作实际上被调度机制隐式打断了。这就是单线程并发中“丢失更新”的典型来源读和写之间出现了检查点。所以命名空间和协程必须放在一起理解。前者决定“读的是哪份数据”后者决定“读的时刻会不会被插入其他写操作”。两者缺一都无法解释很多并发 Bug 的根因。2. Python 命名空间详解LEGB、闭包与声明2.1 查找规则不只是四个字母Python 的名字查找遵循 LEGB 顺序局部作用域Local、闭包作用域Enclosing、全局作用域Global、内建作用域Built-in。每级作用域实际上对应一个或多个命名空间。我整理了一张简表方便对照实际编码时的生命周期层级什么时候生效生命周期局部函数或协程调用时创建帧函数返回或异常时销毁协程挂起时保留闭包嵌套函数引用了外层变量只要闭包对象还存活就一直存在全局模块加载时创建名字表整个模块运行期间内建解释器启动时加载进程生命周期用一个例子说明查找顺序x global def outer(): x outer def inner(): x inner print(x) # inner inner() outer()inner()打印inner不会跳过内层直接读outer。如果注释掉inner里的赋值才会继续查找外层outer作用域找到outer。LEGB 的要点不是背顺序而是知道“找到即停”。每个作用域都可能存在同名变量名字解析永远从最内层开始向外扩散。对协程来说这个顺序完全一样所以很容易下意识地用“调用栈”去理解协程嵌套。但要注意作用域是词法决定的由定义代码的位置决定不是由调用链决定的。也就是说一个async def里能访问哪些外层变量写代码时已经固定了和它在运行时被谁调用没有任何关系。2.2 闭包延迟绑定协程里的经典坑闭包是很容易出问题的场景因为闭包捕获的是名字不是值。看这个协程例子import asyncio async def run(): tasks [] for i in range(5): async def worker(): await asyncio.sleep(0.01) return i tasks.append(worker()) return await asyncio.gather(*tasks) print(asyncio.run(run())) # 输出结果大概率是 [4, 4, 4, 4, 4]为什么不是[0, 1, 2, 3, 4]因为worker闭包里引用的i是run函数局部命名空间中的那个名字。循环每次迭代只是重新给同一个i赋值并没有为每次迭代创建新的作用域。循环结束后i的值停留在 4所有worker真正执行时读到的是同一个最终值。普通同步代码里这个坑同样存在但异步代码里更容易踩中协程被创建到被调度之间往往隔着好几个事件循环轮转创建时的值早就变了。解决办法有两种。第一种是利用默认参数在定义时绑定值for i in range(5): async def worker(ii): await asyncio.sleep(0.01) return i tasks.append(worker())默认参数在定义时求值相当于把当时的i复制到了worker自己的局部命名空间里。第二种是引入一个工厂函数def make_worker(i): async def worker(): await asyncio.sleep(0.01) return i return worker() for i in range(5): tasks.append(make_worker(i))工厂函数每次调用都会创建独立的i命名空间闭包捕获的是各自作用域里的名字逻辑更清晰。我更推荐第二种因为它不会让默认参数承担额外的状态绑定职责代码读起来也直观。2.3 global 和 nonlocal写法、误用与协程场景global x告诉编译器在这个函数里出现的x对应模块级命名空间中的那个名字。nonlocal x则告诉编译器这个x不属于当前函数而是属于外层嵌套函数的作用域。两者的共同点是都是对“名字解析位置”的声明。最容易踩的坑是这样counter 0 def bad_increment(): counter counter 1 # UnboundLocalError看上去只是“读了一下全局再写回去”Python 却报局部变量未定义。原因在于 Python 在编译函数时会做静态分析只要函数体某处存在对counter的赋值整个函数体内的counter都被视为局部变量。于是counter counter 1右侧读取的是那个尚未绑定的局部名字而不是全局名字。修法很明确加一行global counter。在嵌套协程场景下nonlocal也有类似的坑。假设外层函数维护了一个total内层协程要累加它def outer(): total 0 async def helper(): total 1 # UnboundLocalError await asyncio.sleep(0) ...total 1在内层函数里同样被当成局部赋值必须先声明nonlocal total。区别在于nonlocal绑定的是词法作用域中最近的外层定义不是“调用者”的作用域。协程嵌套层数一多有人会误以为nonlocal能跳到调用链上的某个上下文这是不对的。它只认定义处的词法位置。另外global声明在协程里也需要注意你写global shared之后所有协程修改的都是同一份模块级对象。如果没想清楚共享范围这通常就是状态污染的开始。2.4 命名空间的创建与销毁时机普通函数每次调用都会创建新的局部命名空间函数执行完整个命名空间销毁局部数据随之消失。协程不一样async def被调用后返回一个协程对象函数体并没有立即执行协程停在哪个await上局部命名空间就保留在帧对象中等待事件循环恢复它。所以协程的局部变量天然具备“跨挂起存活”的能力这也是状态机、流式处理等模式能在协程里优雅实现的原因。当一个协程被取消或抛出未捕获异常时它的帧会被销毁局部命名空间随之回收。很多资源泄漏问题本质就是协程一直挂在某处没有及时销毁。后面谈取消与超时时会再展开。3. 协程底层机制与事件循环3.1 生成器最初的协程载体Python 协程的根源其实在生成器。生成器函数执行到yield时会暂停整个执行帧和局部命名空间都被保留下来之后用next()或send()恢复它就能接着之前的执行状态继续跑。这天然就是一种协程函数可以主动让出执行权并在之后从让出的位置恢复。看一下这个生成器状态机def counter(): total 0 while True: received yield total if received: total received每次yield total把当前累加值提交给外部外部通过send()传回一个数生成器继续累加。这里的total不会因为yield而丢失它保存在生成器的帧命名空间里。后来async/await出现之前不少异步框架就是靠生成器的yield实现手动协程调度。理解这段历史有助于你明白async def里那些局部变量为什么会跨越await存活本质上和生成器一样都在帧里保留状态。3.2 async/await语法糖背后的执行模型async def是生成器协程的进化版。调用async def函数不会执行函数体而是返回一个协程对象必须把它交给事件循环才能运行。await等待一个可等待对象同时交出控制权asyncio.create_task()则把协程封装成独立任务实现并发。事件循环维护一个就绪任务队列。它反复执行以下流程从队列取一个任务让协程继续执行直到遇到await该协程进入等待状态循环再取下一个任务。由于是单线程任何一段不含await的纯计算代码一旦开始就必须跑完不会被其他协程中途打断。这个特性至关重要它决定了你的代码能在哪里安全地访问共享状态。看一个简单的调度示例import asyncio async def say(name, delay): await asyncio.sleep(delay) print(name) async def main(): await asyncio.create_task(say(A, 0.2)) await asyncio.create_task(say(B, 0.1)) asyncio.run(main())create_task让两个协程并发运行。A 先被调度遇到sleep(0.2)挂起B 被调度遇到sleep(0.1)挂起事件循环等待最短的定时器先唤醒 B打印B再唤醒 A打印A。这就是事件循环调度模型的迷你演示。3.3 await 即检查点单线程如何产生并发交错理解“检查点”比理解await语法本身更重要。任何await都是一个潜在的上下文切换点哪怕await asyncio.sleep(0)这种看起来“什么都没做”的代码也会把控制权交给事件循环。于是下面这个例子会丢更新import asyncio total 0 async def increment(): global total current total await asyncio.sleep(0) total current 1 async def main(): await asyncio.gather(*(increment() for _ in range(10))) print(total) # 大概率不是 10而是 1 或接近 1 asyncio.run(main())十个协程基本都在同一轮sleep(0)之前读取了total 0各自挂起恢复后每个协程把0 1写回total最终结果被覆盖多次。去掉await asyncio.sleep(0)情况就会不同因为每个increment里没有检查点事件循环不会在它运行途中插入其他任务整个读改写过程是原子的。虽然这只是单线程的“原子性”并不是真正意义的并发安全但它揭示了核心原则共享状态跨越 await 的读改写需要额外保护。4. 用命名空间视角设计并发架构4.1 最小可见性原则我写并发代码时最看重的一条原则叫“最小可见性”状态尽量放在尽可能内层的作用域里。常量配置放模块级用大写命名运行时可变的状态放协程局部多个协程之间必须共享的数据放进队列、锁或contextvars等明确设施而不是直接丢给全局变量。原因很简单局部命名空间越窄影响范围越小排障时越容易定位。如果把状态放在模块级整个模块任何位置都能隐式修改它系统规模一大你很难说清某个变量为什么变了。放局部或显式共享设施里数据流方向一目了然。一个可操作的判断标准是当你给变量起名字时问自己“这个变量需要活多久需要被谁看到”如果答案只是“当前协程内部”就别提升它到作用域外层。如果你发现变量被多个协程写第一反应不应该是“加锁”而是“能不能让每个协程各自维护副本最后汇总”。4.2 contextvars并发环境下的请求级命名空间asyncio单线程并发下最怕把“请求 ID”“用户身份”“事务编号”这类上下文信息塞进全局变量。因为多个请求交错执行全局变量很容易互相污染。标准库给出的解法是contextvars.ContextVar它维护一个“上下文”中的变量副本。在 asyncio 中每个任务在创建时会复制当前上下文任务内部修改上下文变量只影响自己和由自己创建的子任务。import asyncio from contextvars import ContextVar req_id ContextVar(req_id, default-) async def child(tag: str): print(f{tag}: {req_id.get()}) async def handle(r): req_id.set(r) await asyncio.gather(child(a), child(b)) async def main(): await asyncio.gather(handle(r1), handle(r2)) asyncio.run(main())handle(r1)设置了自己的上下文版本child(a)和child(b)在创建时复制了这个上下文所以读到r1handle(r2)的任务同样是独立副本读到r2。这相当于给每个请求画了一条独立的命名空间边界。高并发服务里做全链路日志追踪时这个特性非常好用比手动把参数层层传下去干净得多。4.3 完整示例一个带背压的异步任务调度器现在把这些思路拼成一个可运行的小系统。它包含生产者、消费者和信号量限制展示命名空间如何分层、队列如何传数据、contextvars如何保留请求上下文。import asyncio from contextvars import ContextVar from dataclasses import dataclass request_id_ctx ContextVar(request_id, default-) dataclass class TaskItem: url: str retry: int 0 dataclass class TaskResult: url: str status: int body_size: int request_id: str error: str async def fetch(url: str) - tuple[int, int]: # 模拟网络 IO真实项目可换成 aiohttp / httpx await asyncio.sleep(0.02) return 200, len(url) * 10 async def worker(wid, in_q, out_q, sem): while True: try: item await in_q.get() except asyncio.CancelledError: break # 每个请求的上下文在这里显式设置只影响当前 worker 任务 request_id_ctx.set(fworker-{wid}-{item.url}) try: async with sem: status, size await fetch(item.url) result TaskResult(item.url, status, size, request_id_ctx.get()) except Exception as exc: result TaskResult(item.url, 0, 0, request_id_ctx.get(), errorstr(exc)) finally: out_q.put_nowait(result) in_q.task_done() async def producer(urls, in_q): for url in urls: # 队列满时这里挂起生产者不会无限生产这就是背压 await in_q.put(TaskItem(url)) async def main(): urls [fhttps://example.com/{i} for i in range(30)] in_q asyncio.Queue(maxsize5) out_q asyncio.Queue() sem asyncio.Semaphore(5) workers [ asyncio.create_task(worker(i, in_q, out_q, sem)) for i in range(4) ] producer_task asyncio.create_task(producer(urls, in_q)) await producer_task await in_q.join() for w in workers: w.cancel() await asyncio.gather(*workers, return_exceptionsTrue) results [] while not out_q.empty(): results.append(out_q.get_nowait()) print(f处理完成 {len(results)} 个任务) for r in results[:5]: print(f{r.url} status{r.status} size{r.body_size} ctx{r.request_id}) asyncio.run(main())这个骨架有几个值得复用的设计点。队列是生产者与消费者之间的唯一数据通道worker 拿到了数据各自处理不需要共享任何可写对象Semaphore把同时进行的网络请求限制在 5 个以内避免打爆下游in_q.maxsize5让生产者遇到队列满时自动暂停从而实现背压任务不是一次性全部涌入内存。request_id_ctx的设置位置也很讲究每个 worker 在自己的任务上下文中设置不会影响其他 worker因为任务创建时复制上下文运行时彼此独立。同时在finally里用out_q.put_nowait(result)即使请求异常也会产出结果不会丢任务。4.4 用队列传递数据而不是共享对象写了这么久并发代码我的体会是“队列”这个词的价值被人严重低估了。很多人一遇到共享状态就想到锁但锁解决的是“临界区互斥”并没有解决“数据流混乱”。队列强制规定了数据流向只有一个协程产生另一个协程消费方向明确命名空间边界也随之清晰。能用队列沟通就别让协程之间直接写彼此的局部变量。局部变量本来属于单个协程的私有命名空间跨协程访问会破坏封装。如果你想让某个协程对其他协程发布状态最稳妥的方式是把状态放到消息里通过队列广播或投递。这看起来绕实际上把并发控制从“隐式共享”变成了“显式通信”排障和测试都会轻松很多。5. 高频问题排查与性能调优实录5.1 UnboundLocalError隐式局部变量与作用域陷阱这个问题太常见了值得单独拿出来说。之前提到函数体内任何位置出现对名字的赋值这个名字就会在整个函数体内被当成局部变量。协程函数也一样因为async def也是函数局部变量判定规则完全相同。你可能遇到的情况是模块里有个config协程里第一行print(config)正常第二行config ...却报了UnboundLocalError。原因就是后面那个赋值导致整个协程里的config都变成局部作用域前面读取的操作反而找不到绑定。排查时只要把赋值去掉或改成局部变量问题就消失。这一点虽然简单但在深层嵌套的并发代码里很容易被忽略。遇到这类错误先看函数体里有没有同名赋值。5.2 丢失更新读改写之间不要放 await计数器、缓冲水位、自增 ID这类“读改写”操作在异步代码里特别容易丢更新。前面已经演示过await是调度点一旦读和写之间隔着await其他协程就可能插入修改。这里的建议是如果没有特别的并发要求避免用模块级共享变量累加。改成每个协程局部统计最后聚合。如果必须共享用asyncio.Lock把整个读改写包起来但要注意等锁本身就是检查点所以进入临界区后要重新读取状态。大量任务同时写一个共享对象时考虑换成分片命名空间比如一个列表分成多个桶每个协程写自己的桶最后汇总。这个技巧在高频计数场景下非常有效。async with lock: total 1请注意上面这行代码获取锁之后total 1会重新读取total所以不会出现“拿着旧状态写新状态”的情况。如果你把total读取放在async with外面即使加了锁也可能丢更新。5.3 CPU 密集任务不能阻塞事件循环await是协程让出控制权的地方但如果一个协程里运行了耗 CPU 的纯计算中间没有任何await事件循环只能等它跑完其他协程全部卡住。这不是并发问题而是调度饥饿。处理办法是把这类计算放到独立线程或进程比如asyncio.to_thread()处理阻塞 IOProcessPoolExecutor处理重 CPU 任务。这里也能看出命名空间边界的意义进程池里的子进程不共享父进程的模块级命名空间数据要明确地通过返回值或队列传回来。把 CPU 任务拆到子进程实际上就是利用了进程隔离的命名空间避免共享状态冲突。很多高并发系统的性能瓶颈不是 Python 解释器慢而是把 CPU 密集段放在了事件循环线程里。5.4 超时、取消与任务异常的正确处理asyncio.wait_for(coro, timeout)是限制任务耗时的常规写法try: result await asyncio.wait_for(fetch(url), timeout2.0) except asyncio.TimeoutError: # 超时处理但要注意超时之后内部的协程会被cancel()。如果你的协程内部使用了资源比如打开连接、占用信号量最好在finally里做清理否则会留下悬挂状态。更隐蔽的是任务异常没有处理时的现象创建了任务但不await它协程内部抛出的异常很可能不会显式弹出而是由事件循环在日志里提示Task exception was never retrieved。这会延迟报错排查很有迷惑性。推荐的做法是显式await或给任务添加done回调至少确保异常能被检索。取消任务时还有一个容易引发二次异常的小坑如果在await in_q.get()这里被取消代码里的finally in_q.task_done()就可能访问一个未绑定的item反而触发新的异常。所以我在上面的调度器里把get()单独放进一层try捕获CancelledError后主动break把清理逻辑和队列获取解耦。5.5 用日志和调试模式量化协程调度调高并发系统光靠 print 不够最好借助 asyncio 自带的调试能力。设置环境变量PYTHONASYNCIODEBUG1运行程序事件循环会输出更详细的警告比如未检索的任务异常、回调耗时过长等。还可以在代码里主动度量任务数量loop asyncio.get_running_loop() print(f活动任务数: {len(asyncio.all_tasks())})在关键节点打点统计任务数、队列长度和平均处理耗时能帮你判断瓶颈在哪。比如in_q.size()长期处于 0说明生产者跟不上sem._value长期很小说明并发限制太紧任务数暴涨且回收慢多半存在任务泄漏创建了不带超时控制的长生命周期协程。最后说一个我自己反复踩过的经验遇到奇怪的并发问题先别急着加锁。把涉及共享状态的变量逐个列出来问一句“它到底属于哪一层作用域如果多个协程都会修改它是否有必要做全局共享”如果这个问题的答案需要犹豫那大概率就是命名空间边界没画清楚。把这条准绳记住再配合await检查点意识并发代码出错的概率会小很多。