1. 并发与并行先搞清楚你面对的到底是哪个问题聊Python并发十个有九个半会先撞上GIL这堵墙。但很多新手还没走到GIL那一步就已经把并发和并行两个词混着用了。先说人话版本并发是多个任务在同一个时间段内交替推进看起来像同时进行并行是多个任务真正在同一时刻各自跑在不同的CPU核心上。前者是一个人分时做几件事后者是几个人同时各干各的。Python里这两个概念落地之后区别非常明显。因为GIL的存在纯Python代码的多线程并行基本是被限制的——同一时刻只有一个线程能执行字节码。但并发不受影响线程可以交替执行遇到I/O操作时主动让出解释器这就是为什么Python多线程处理网络请求、文件读写这类I/O密集型任务时依然能打但面对纯CPU计算密集型任务时多线程反而可能因为锁竞争比单线程还慢。我自己刚接触这块时的状态特别典型以为开了线程池就万事大吉结果一跑CPU密集的循环8个线程加起来比单线程还慢当时整个人是懵的。后来才明白Python并发方案的选择从来不是哪个更高级而是你的瓶颈到底卡在哪儿。判断标准其实很简单——你的程序是在等I/O还是等CPU。等I/O的用多线程或asyncio等CPU的用多进程或直接拥抱C扩展。这篇文章我会从GIL的工作原理开始讲然后掰开揉碎分析threading、multiprocessing、asyncio三大方案各自的适用场景和踩坑点最后给出一套高并发实战的完整思路。不管你是写爬虫、做Web服务还是搞AI Agent的并发调度这套思路都应该能帮你少走不少弯路。2. GIL的核心机制为什么你的多线程总是差点意思2.1 GIL到底是什么它管的究竟是谁GIL全称Global Interpreter Lock全局解释器锁。它是CPython解释器的一个核心设计——注意是CPythonPyPy和Jython这类实现并不受这个限制。GIL的职责是保证同一时刻只有一个线程在解释器层面执行Python字节码。换句话说你开了10个线程在纯计算场景下这10个线程不会真正同时跑在10个CPU核上而是轮流用同一个核严格说是同一把锁操作系统依然会把线程调度到不同核上但只有拿到GIL的线程才能干活。这带来的直接后果就是Python多线程的CPU密集任务性能不仅不会提升还可能因为线程切换开销、锁获取竞争比单线程还要差。但这里有个关键点需要说透GIL锁住的是Python字节码的执行不是所有的操作。C扩展模块、I/O系统调用、部分标准库的底层实现在执行时是可以释放GIL的。比如time.sleep()、文件读写、socket通信这类操作在等待期间线程会主动释放GIL让其他线程拿到执行权。这正是多线程在I/O密集型任务中依然有效的根本原因。2.2 为什么GIL迟迟不消失以及它对性能的真实影响很多人问过一个问题GIL这么碍事为啥不干脆去掉答案其实比想象中复杂。GIL的存在换来的是CPython实现的大幅简化——内存管理不需要考虑多线程同时修改对象引用的复杂情况引用计数在GIL的保护下变得非常安全。去掉GIL意味着整个解释器的内存模型要重新设计还要保证所有现有C扩展库的兼容性这个工程量超乎想象。Python官方一直在推进no-GIL的构建方案比如PEP 703但短期内主流CPython发行版依然会有GIL。拿一个真实的例子来感受一下GIL的影响。假设我有四段纯计算的任务每段运行大约1秒import threading import time def compute(rounds): total 0 for i in range(rounds): total i ** 2 return total # 单线程串行 start time.time() for _ in range(4): compute(50_000_000) print(f单线程耗时: {time.time() - start:.2f}s) # 多线程 threads [] start time.time() for _ in range(4): t threading.Thread(targetcompute, args(50_000_000,)) threads.append(t) t.start() for t in threads: t.join() print(f多线程耗时: {time.time() - start:.2f}s)实测下来多线程版本不仅没有比单线程快反而会因为上下文切换和GIL竞争多出20%到30%的开销。很多初学者在这里容易得出多线程无用的结论这是不对的——准确的结论应该是多线程不适合Python的CPU密集计算场景。2.3 GIL在I/O密集场景下的真实表现把上一节的计算任务换成网络请求情况会完全不同。假设你需要并发请求100个HTTP接口每个请求平均耗时200毫秒。用多线程实现100个线程同时发出请求线程在等待响应的过程中都会释放GIL所以100个请求的总耗时大约是200到300毫秒取决于连接池和调度开销而不是单线程的20秒。这就是GIL机制的微妙之处它锁的是执行不锁等待。I/O等待期间线程根本不需要GIL所以Python多线程在I/O密集场景下能发挥出接近真正的并行效果。我自己写爬虫和API调用服务时基本都是用线程池来处理配合requests.Session或aiohttp连接复用效果非常稳定。如果你在做一个需要同时请求大量外部API的服务threading模块配合concurrent.futures是最容易上手的方案。3. 三大并发方案选型threading、asyncio、multiprocessing到底怎么挑3.1 threading简单直接但别踩线程安全的坑threading模块是Python并发入门的标配。它的心智模型最接近操作系统的线程概念用Thread类创建线程用join等待线程结束用Lock、Semaphore、Event等同步原语控制线程间协作。一个最容易犯的错误是直接操作共享变量而不加锁。比如下面这个经典例子import threading counter 0 def increment(): global counter for _ in range(1_000_000): counter 1 threads [] for _ in range(5): t threading.Thread(targetincrement) threads.append(t) t.start() for t in threads: t.join() print(counter) # 结果不是5_000_000而是比这个小的随机数这个问题的本质是counter 1不是原子操作它分为读取、计算、写回三步。两个线程同时读取到同一个counter值各自加1后再写回就会丢失一次更新。这种情况必须加锁lock threading.Lock() def increment(): global counter for _ in range(1_000_000): with lock: counter 1但加锁会带来性能损耗。如果任务是纯计数器累加这类操作还可以换成threading.local()把变量变成线程私有或者用queue.Queue做线程间通信——队列本身自带锁机制既安全又省心。我的习惯是能不用共享状态就不用能塞进队列就让线程从队列取任务、向队列写结果这种做法出错概率最低。3.2 asyncio单线程的事件循环能扛万级并发asyncio是Python 3.4引入的异步I/O框架它的并发模型和线程完全不同。asyncio在单个线程内维护一个事件循环通过await挂起当前任务切换到其他任务执行。这里的核心是协程coroutine——一个可以用await暂停和恢复的函数。用一句话总结asyncio的适用场景大量I/O等待、但I/O之间没有太多CPU计算的任务。典型例子是并发抓取大量URL、WebSocket长连接服务器、消息队列消费者等。来看一个最简单的asyncio并发抓取的示例import asyncio import aiohttp async def fetch(session, url): async with session.get(url) as response: return await response.text() async def main(): urls [fhttps://httpbin.org/get?n{i} for i in range(100)] async with aiohttp.ClientSession() as session: tasks [fetch(session, url) for url in urls] results await asyncio.gather(*tasks) print(f抓取完成共{len(results)}个响应) asyncio.run(main())这段代码里asyncio.gather负责并发调度。100个网络请求几乎同时发出单线程内的事件循环在等待响应的间隙会切换到其他协程执行。实测100个请求总耗时大约就是最慢那一个请求的耗时比串行快一到两个数量级。asyncio和threading的本质区别在于协作式调度。asyncio是任务自己主动让出执行权遇到awaitthreading是操作系统强制抢占。所以asyncio的并发效率更高没有线程上下文切换但要求代码里不能出现阻塞调用——一旦某个协程里用了time.sleep()或者同步的requests.get()整个事件循环都会卡住其他协程全部停摆。这是新手最容易踩的坑协程内部不小心用了同步I/O库程序直接变串行。我建议的选型标准是并发任务数量非常大成千上万且都是I/O密集型优先asyncio任务数量几百个、需要和旧代码集成紧密、或者涉及复杂的线程安全交互优先线程池。3.3 multiprocessing绕过GIL的终极方案multiprocessing模块是Python应对CPU密集型任务的方案。每个进程有自己独立的Python解释器和内存空间没有GIL争抢问题。多进程更适合真正的并行计算——把一个大任务分解成多个独立子任务分派给多个CPU核心同时计算。使用方法很简单from multiprocessing import Pool def square(x): return x * x if __name__ __main__: with Pool(processes4) as pool: results pool.map(square, range(100)) print(results)Pool.map会把任务自动分配到4个进程执行对于纯计算任务效果接近线性的性能提升前提是任务本身可以并行分解。不过multiprocessing有两个麻烦事一是进程间通信不能直接共享变量需要借助Queue、Pipe或者共享内存二是进程启动开销大不适合轻量级短任务。还有一个坑是Windows下multiprocessing需要把入口代码放进ifname main保护块中否则会递归启动子进程导致报错。macOS上如果进程启动方式不对也可能引发冻结问题。这些细节虽然小但遇到时排查成本不低。3.4 选型对照一张表理清思路方案适用场景CPU密集I/O密集共享状态学习成本典型示例threadingI/O密集、需要线程间交互差GIL限制好需要加锁/队列低爬虫、API并发请求asyncioI/O密集、高并发连接差单线程执行极好无需共享协程内隔离中WebSocket服务、高并发HTTPmultiprocessingCPU密集计算极好一般需要进程间通信中高数据批量处理、图像/数值计算4. 高并发实战从零搭一个可支撑高并发的IM后端4.1 场景描述与架构决策先抛一个要实战的方向——我之前做过一个类似于企业内部沟通工具的IM后端服务场景是1万个的同时在线连接每条消息要在100毫秒内送达对方还要求消息不丢失、不重复。这个场景下WebSocket长连接是最合适的选择因为IM需要服务端主动推送消息给客户端HTTP轮询在这种场景下效率太低。技术选型时我首选asyncio FastAPI WebSocket。FastAPI最迟从0.95版本开始支持原生WebSocket底层走的是Starlette uvicorn整个链路是异步的非常适合长连接高并发场景。Python的asyncio单线程事件循环理论上可以扛几万个WebSocket长连接——瓶颈其实根本不在CPU而在文件描述符的数量和网络带宽上。4.2 服务端核心实现连接管理与消息分发连接管理是IM后端最关键的设计问题。每个WebSocket连接对应一个用户服务端必须快速找到某个用户对应的连接才能把消息正确推送过去。我用一个全局字典来维护用户ID到WebSocket对象的映射配合asyncio.Lock保证并发安全import asyncio from fastapi import FastAPI, WebSocket, WebSocketDisconnect app FastAPI() connections {} connections_lock asyncio.Lock() app.websocket(/ws/{user_id}) async def websocket_endpoint(websocket: WebSocket, user_id: str): await websocket.accept() async with connections_lock: connections[user_id] websocket try: while True: data await websocket.receive_text() # 解析消息、路由、持久化 await handle_message(user_id, data) except WebSocketDisconnect: async with connections_lock: if connections.get(user_id) websocket: del connections[user_id] async def send_to_user(target_user_id: str, message: str): async with connections_lock: ws connections.get(target_user_id) if ws: await ws.send_text(message)这里有个关键细节为什么操作connections字典要加锁因为asyncio是单线程事件循环理论上同一个时刻只执行一个协程字典操作看起来不会冲突。但await语句会把控制权交还给事件循环在await期间可能有其他协程修改了connections。用一个锁把查找连接和发送消息包起来才能保证不会被其他协程打扰。这种锁保护的读改写模式在asyncio里非常重要。4.3 优化消息推送的下游从主动推送到发布订阅上面直接发送的方式在连接数少时够用但1万连接时向一个用户推送消息涉及字典查找和WebSocket写入如果单线程逐个推送耗时不可控。更好的做法是引入一个消息队列或内存中的发布订阅中心核心思路是当有一条消息进来时把它抛给分发器分发器根据目标用户ID找到对应连接把消息写进去。这里我用asyncio.Queue做这一步把接收消息和发送消息解耦message_queue asyncio.Queue() async def worker(): while True: target_user_id, content await message_queue.get() await send_to_user(target_user_id, content) message_queue.task_done() app.on_event(startup) async def startup(): for _ in range(10): # 开启10个消费协程 asyncio.create_task(worker())10个消费协程同时从队列取消息配合asyncio的并发调度消息分发的吞吐会明显好于单协程逐个推送。这个设计本质上是生产者-消费者模型消费协程的数量可以根据消息量动态调整。4.4 再压一层消息持久化与幂等策略IM场景下数据不丢失、消息不重复是基本要求。实测中最省心的方案是把消息先写入Redis或类似的高性能存储做缓存然后异步批量刷入MySQL。这里有个先写缓存、再异步落库的思路可以避免每条消息都即时同步写MySQL造成性能瓶颈。幂等策略也必须在设计时考虑清楚。最简单的方式是每条消息带上由发送方ID客户端时间戳随机数组成的消息ID服务端用Redis的SETNX检查这个消息ID是否已存在存在则拒绝不存在则处理并在Redis里留下标记。这样可以保证即使客户端重试、或服务端在写库前崩溃了恢复后重放消息也不会产生重复数据。async def is_duplicate_message(message_id: str) - bool: result await redis.set(fmsg:{message_id}, 1, nxTrue, ex600) return result is None上面这段是借助Redis的NX选项实现只允许首次写入配合过期时间防止标记无限积累。整个过程完全异步在高并发下性能表现很好。4.5 客户端侧压力测试看服务端能扛到多少完成上述代码后需要做压测。压测工具我用的是websocket-bench和Locust。测试方案是模拟5000、10000并发连接每个连接每5秒发一条消息记录服务端的CPU、内存、延迟和消息处理吞吐。实测一个关键现象当连接数突破1万时服务端CPU占用率并没有急剧上升但操作系统的文件描述符限制会先成为瓶颈。Linux默认的ulimit -n是1024意味着默认情况下一个进程最多只能打开1024个文件1万个WebSocket连接完全跑不起来。所以压测前必须调大系统级和进程级的文件描述符限制# 查看当前限制 ulimit -n # 临时调整 ulimit -n 65535 # 永久调整修改 /etc/security/limits.conf # * soft nofile 65535 # * hard nofile 65535压测结果很能说明问题在8核16G的云服务器上asyncio方案可以稳定支撑1.2万左右的并发WebSocket连接每条消息的平均延迟在30到50毫秒之间。CPU占用率大约维持在60%到70%。如果是用多线程方案做同样的连接管理在相同配置下大约只能撑到2000到3000个连接而且延迟波动明显更大。这就是事件循环模型在I/O密集型场景下的优势。5. 常见问题与排查技巧实录5.1 GIL相关多线程比单线程还慢我最早遇到这个问题时排查了很久最后发现是因为代码里有大量的循环计数和字符串拼接操作这些都是纯Python字节码完全被GIL约束。排查方法很简单在代码里加入cProfile看CPU时间集中在哪些函数上。如果集中在纯Python代码段那直接换multiprocessing如果集中在I/O等待上说明多线程是合适的。另外还有一个容易被忽略的坑有些C扩展库在调用时不会释放GIL比如一些加密库、图像处理库的Python绑定实现。即使你开了多线程这些库的调用过程仍然会阻塞其他线程的执行。排查这类问题时可以用perf工具或py-spy来采样线程栈看看线程到底卡在哪儿。5.2 asyncio协程卡死不执行asyncio最经典的问题场景某个协程里调用了time.sleep()或同步的requests.get()导致事件循环被阻塞其他协程全部卡死。这是asyncio新手最容易犯的错误。解决方案分几层如果必须用同步库就用asyncio.to_thread把阻塞操作丢到独立线程里执行或者直接换异步库比如requests换成aiohttp。我自己的习惯是严格区分协程内的库和协程外的库绝不在协程内使用同步阻塞调用。另一个常见问题是忘记把协程对象包装成Task。直接在事件循环中调用async函数不会自动并发执行没有await的话协程根本不会运行。正确写法是使用asyncio.create_task()创建后台任务或者用asyncio.gather()收集所有协程并发执行。5.3 线程池任务卡死、队列越积越多线程池配合Queue用的时候如果任务队列一直增长但消费速度跟不上通常是线程数量设置不合理或者某个任务内部有长期阻塞的I/O操作。排查思路是先统计单个任务的平均执行时间再根据期望的吞吐量反推线程数。公式很简单线程数约等于每秒期望任务数乘以单任务耗时秒。如果单任务耗时太长比如超过1秒就要考虑任务拆分或换用asyncio方案。另外线程池里如果用lock不当心容易出现死锁。比如线程A持锁后等待线程B完成某个操作而线程B也在等待线程A释放锁。这种死锁一旦出现进程会完全卡住。排查方法是抓线程栈使用py-spy或faulthandler看各个线程卡在哪些锁上。5.4 问题排查速查表现象可能原因优先排查项多线程CPU密集无提升GIL限制换成multiprocessingasyncio协程排队执行协程内存在同步阻塞调用检查time.sleep、requests、标准库阻塞I/O多线程共享变量值错乱多线程同时读写共享状态加锁或改用queue.Queue进程池启动失败没有main保护块加上ifname main并发连接数不够文件描述符限制检查ulimit -n、系统级limits.conf高并发下延迟抖动事件循环被某任务阻塞用py-spy采样定位卡住的协程5.5 一个压箱底的排查技巧可视化线程与协程调度排查并发问题最忌讳瞎猜。我个人最常用的是这三个工具py-spy可以采样运行中的Python进程的线程栈不带侵入性适合排查卡住的线程。faulthandler在程序崩溃或卡死时打印所有线程的堆栈尤其适合FastAPI、uvicorn这类后台服务。asyncio的debug模式设置PYTHONASYNCIODEBUG1后事件循环会打印协程的执行情况和卡住的I/O操作。但注意debug模式会显著降低性能正式环境不要开。这三个工具配合使用能定位到协程或线程具体阻塞在哪一行代码、哪个函数、哪个锁上比起盲目的print大法高效得多。6. 延伸AI Agent场景下的高并发怎么扛6.1 Agent服务为什么和高并发八字不合最近AI Agent相关的项目越来越多热词里也出现了类似AI Agent怎么扛并发的讨论。Agent服务和传统Web服务在并发处理上的差异非常大传统Web请求是无状态的请求进来、处理、返回就结束了Agent服务往往是有状态的一个请求可能对应一个多轮对话的工作流中间要调用大模型API、工具调用、数据库查询整个流程可能是分钟级甚至更长。更麻烦的是Agent工作流中大部分时间都在等待外部API响应LLM调用、检索、工具执行这些等待时间占整个请求耗时的90%以上。这意味着Agent服务本质上是重I/O的多线程或asyncio反而非常适合——关键是要把一次对话抽象成一个可以异步执行的工作流任务。如果Agent工作流里全是同步阻塞调用并发能力会被拉低好几个数量级。6.2 Agent并发方案把工作流拆成可异步执行的任务我实际设计的Agent后端是这样的用asyncio来处理工作流中的等待型操作调用LLM API、等待工具返回同时用线程池来处理那些难以异步化的同步库调用比如某些数据库驱动、本地工具。工作流本身被建模为一个协程链每个步骤返回一个可等待对象由主事件循环调度。如果某个Agent任务的计算量太大比如要做本地向量检索、重排序就把它丢到multiprocessing池里避免阻塞事件循环。整体架构可以描述为三层接入层用asyncio管理WebSocket/HTTP长连接工作流层用协程编排Agent的多步操作计算层用进程池执行重的CPU计算。这种混合架构能最大程度发挥Python并发模型各自的优势——这是单用某一方案很难达到的效果。6.3 一个亲测有效的Agent请求合并技巧高并发下Agent服务有个常见的性能杀手大量请求在做相同或相似的LLM调用。比如100个用户同时问帮我总结今天的日程每个请求都向大模型发一次几乎一样的Prompt既浪费又加塞。一个实用性很强的优化是请求合并request coalescing在短时间内把相同意图参数的请求合并成一个外部调用外部返回后把结果广播给所有等待的请求。实现思路是pending_requests {} lock asyncio.Lock() async def get_llm_response(prompt: str): async with lock: if prompt in pending_requests: # 已有请求在飞直接等待结果 future pending_requests[prompt] else: # 新请求创建future并发起真正的调用 future asyncio.get_running_loop().create_future() pending_requests[prompt] future asyncio.create_task(execute_llm_call(prompt, future)) response await future # 清理 async with lock: pending_requests.pop(prompt, None) return response这个方案的好处是在同一小段时间内同样的Prompt只触发一次真正的LLM调用所有等待方共享结果。实测在并发峰值从300个请求降到了40个外部调用。坏处是引入了复杂度必须处理超时、异常、future未完成的清理等边界情况。但对于调用成本昂贵、QPS限制严格的LLM服务来说这个优化几乎是值得做的。6.4 Agent并发的一个血泪教训别让状态污染请求Agent工作流中最大的隐藏坑是会话状态串了。如果用全局变量存储某个用户的工作流状态高并发下必然出现状态错乱——用户A的会话上下文跑到用户B那去了。这类问题的终极解法是状态必须跟着请求走把状态对象作为协程参数传递绝不用全局变量。每个请求独立实例化一个状态对象天然隔离不需要加锁也永远不会串数据。这一点在处理Agent场景时尤其重要因为Agent工作流本身有大量中间状态当前上下文、工具调用记录、临时变量一旦串了调试成本极高。我见过很多次线上事故的根因不是代码逻辑错了而是并发状态下共享了全局session。所以我的原则是高并发下所有可变状态都绑定到请求级别能不用全局变量就不用。7. 个人经验总结并发编程的几条红线这套系统跑下来我自己总结出几条红线每一条几乎都是用线上事故换来的教训。第一条能不用锁就不用锁。锁是并发问题的根源之一引入锁意味着可能出现死锁、性能下降、调试困难。优先用队列、future、不可变数据这些天然的并发安全工具。第二条GIL不是罪魁祸首选错模型才是。很多人在多线程性能不佳时骂GIL其实是选型出了问题。先判断任务是I/O密集还是CPU密集再决定用threading还是multiprocessing这一步决定了后面所有设计的走向。第三条不要迷信并发数。并发数高不代表性能好能扛得住延迟抖动才算真稳。我在压测时最关注的是P99延迟——即99%的请求在多少毫秒内完成而不是最大并发数这种宣传指标。P99能暴露的是系统在负载峰值下的真实体质。第四条状态隔离是并发安全的第一道防线。任何可能被多个线程或协程共享的可变状态都要在设计阶段就想清楚归属。归请求的状态归请求归全局的用锁保护千万不要模糊地带。最后再分享一个细节写并发代码时尽量让每个线程或协程只做一件独立的事通过队列传递数据和结果。这样代码的结构天然就是生产者-消费者模型不容易出错性能也更容易分析。我后来把IM服务的消息分发、AI Agent的工作流调度都改成了这种模式线上出问题的频率明显下降。从这个角度回看并发编程的核心其实不是那些花哨的API而是对谁拥有什么状态、谁在什么时候等待什么这两件事的清晰认知。