如果你写过爬虫、批量调接口、处理大量文件读取大概率见过一个现象程序跑起来CPU占用不高但就是卡卡在网络响应上、卡在磁盘读写上整个任务排成一条长队一个请求慢就把后面的全部堵死。很多人第一反应是上多线程结果加锁、调线程池一顿操作之后发现卡顿依旧甚至因为线程切换和资源竞争变得更慢。这个问题的根源不在计算量而在绝大多数时间花在了“等”上。这一篇我会把协程怎么解决这类IO密集型任务卡顿讲透从原理到实操从核心语法到常见坑位一次性整明白。1. IO密集型任务为什么天生容易卡1.1 卡顿的真正来源是“等待”IO密集型任务指的是程序大部分时间不是在算而是在等。等网络数据包返回等磁盘写入完成等数据库查询结果回来。这类任务的典型特征是CPU使用率很低但墙钟时间很长。举个例子你写一个脚本去请求100个网页单线程挨个请求每个请求平均耗时200毫秒那总共就是20秒。这20秒里CPU真正干活的累计时间可能不到0.5秒其余19.5秒全在等网络。很多人不理解为什么不用多线程。用多线程确实可以改善但在Python里多线程受制于GIL全局解释器锁同一时刻只有一个线程能执行Python字节码。当线程A在等待网络响应时线程B可以切换过来执行所以多线程在IO密集型任务里确实能提速。但线程的创建和切换有成本线程多了之后光是上下文切换就能吃掉不少CPU而且共享资源的锁管理很容易引入隐蔽bug。更关键的是线程切换是由操作系统抢占式调度的你无法精确控制切换时机这导致并发行为不可预测。协程方案完全换了一个思路不等操作系统来切主动让出。程序遇到IO等待时自己挂起当前任务告诉事件循环“我可以让位了你先跑别的”等IO完成之后再回来继续。这种方式没有线程切换的开销也不需要加锁保护共享数据因为同一时刻只有一个协程在运行。1.2 阻塞与非阻塞问题藏在API里这里有个容易忽略的点同样的IO操作用错了API就会变回阻塞。比如你用requests库请求网页这个库内部是同步阻塞的一旦发出请求当前线程就休眠直到响应到达。即使外面套了协程只要在协程里调用了requests.get整个事件循环一样会被卡住。真实的IO密集任务卡顿绝大多数是因为同步阻塞调用混进了异步流程。要么用了同步请求库要么用了同步的文件读写要么用了同步的数据库驱动。这些都是经验之谈排查协程性能问题时先看IO操作是否用了正确的异步版本八成问题都出在这。2. 协程是怎么做到不卡顿的2.1 事件循环一个人干多个人的活协程的核心机制是事件循环。你可以把事件循环理解成一个调度员它维护一个任务队列不断轮询从队列里取出一个协程任务让这个任务执行到某个await处如果await的东西还没准备好任务就挂起调度员转去执行下一个任务当IO事件完成回调触发调度员把之前挂起的任务重新塞回队列。这样单线程内实现了并发不是时间的并行而是等待的复用。一个人同时向100个URL发出请求然后挨个等他们返回——不是挨个串行地等而是全部发出去后谁先返回就先处理谁。生活化类比去餐厅点餐串行方式是点完坐下等A菜的厨师做好再等B菜全部上齐再吃。协程方式是点完餐先干别的事A菜好了服务员叫你B菜好了再叫你一顿饭下来你把时间都用上了。线程呢相当于雇了俩服务员分别盯A菜和B菜人多了服务员本身就乱。2.2 async/await 语法背后的状态机async def定义的函数不再是普通函数调用时会返回一个协程对象而不是立即执行。await关键字是挂起点它后面的表达式必须是一个可等待对象比如另一个协程、Future或者Task。执行到await时当前协程会暂停把控制权交还给事件循环。这个机制底层是一个状态机。每一个await都是一个状态断点协程对象保存当前的局部变量和执行位置。当await的对象结束后Python会恢复保存的状态从断点继续执行。所以你能在协程里写线性的代码逻辑看起来像同步跑起来是异步。理解这个状态机很关键因为这意味着协程的切换点完全由你代码里的await决定。只要你的代码中没有await这一段就是原子的不会被打断。这既是优势不需要锁也是陷阱如果一段长时间CPU密集操作没有await整个事件循环就被你独占了。3. 实操用asyncio改造真实IO任务3.1 案例背景批量请求公共接口我用一个完整的案例来演示改造过程。场景是给定100个待查询的商品ID需要调用某个公共HTTP接口获取价格信息。这是一个典型的IO密集型任务——不发请求时CPU是空闲的发请求后就是等网络。先看最原始的同步版本import time import requests def fetch_price(product_id): # 模拟一个HTTP请求假设每个请求耗时0.2秒 resp requests.get(fhttps://api.example.com/price/{product_id}, timeout5) return resp.json()[price] def sync_main(): start time.perf_counter() prices [] for i in range(100): prices.append(fetch_price(i)) print(f同步耗时: {time.perf_counter() - start:.2f}s) sync_main()这个版本跑下来大约20秒。所有请求都是一个一个发的每个都等它响应完才发下一个。网络往返时间完全被浪费了。3.2 改成协程版本改造的第一步是把同步的requests库换成异步HTTP客户端。我用httpx库举例因为它同时支持同步和异步API语法上最接近requests迁移成本低。pip install httpx协程版本核心代码如下import asyncio import time import httpx async def fetch_price(client, product_id): resp await client.get(fhttps://api.example.com/price/{product_id}) return resp.json()[price] async def async_main(): async with httpx.AsyncClient(timeout5) as client: tasks [asyncio.create_task(fetch_price(client, i)) for i in range(100)] prices await asyncio.gather(*tasks) print(f协程耗时: {time.perf_counter() - asyncio.get_running_loop().time():.2f}s) # 注意性能计时要用 loop.time() 或 time.perf_counter这里简化为 time 方式这里有个细节我要纠正一下上面代码里的计时方式容易混我直接写清楚async def async_main(): start time.perf_counter() async with httpx.AsyncClient(timeout5) as client: tasks [asyncio.create_task(fetch_price(client, i)) for i in range(100)] prices await asyncio.gather(*tasks) print(f协程耗时: {time.perf_counter() - start:.2f}s)跑完你会发现耗时大约0.3到0.5秒比20秒快了四十倍以上。这就是并发的威力。100个请求同时发出最大的等待时间取决于最慢的那个请求而不是所有请求耗时之和。关键代码解读async with httpx.AsyncClient(...)创建异步客户端它内部维护连接池支持并发复用底层连接asyncio.create_task(coro)把协程包装成Task立即调度到事件循环不阻塞当前代码asyncio.gather(*tasks)等待所有Task完成返回结果列表顺序与传入顺序一致。3.3 控制并发量别把对方服务打死上面例子一次性发出100个请求如果请求数变成10000就会瞬间把目标服务的连接池打满甚至触发对方防火墙的限流机制。协程虽然不耗线程但网络连接本身是资源端口、文件描述符都有上限。所以实际开发中必须要控制并发上限。asyncio提供了Semaphore信号量来实现流量控制import asyncio import httpx sem asyncio.Semaphore(20) async def fetch_with_limit(client, product_id): async with sem: resp await client.get(fhttps://api.example.com/price/{product_id}) return resp.json()[price] async def main(): async with httpx.AsyncClient(timeout5) as client: tasks [asyncio.create_task(fetch_with_limit(client, i)) for i in range(1000)] prices await asyncio.gather(*tasks) print(len(prices))Semaphore(20)的意思是同一时刻最多有20个协程进入临界区。其余协程在async with sem这行等待等有协程释放了信号量才能继续。这种做法在接口调用、数据库批量写入、爬虫里非常常用也是专业项目与Demo之间的分水岭。还有一个容易被新手忽略的点Semaphore必须在事件循环创建之前或之内创建。如果你在模块顶层直接sem asyncio.Semaphore(20)Python 3.10以上会出现警告因为Semaphore绑定到了当前线程的事件循环。正确做法是在协程函数内部创建或者在main协程里创建后作为参数传递。4. 实战中踩过的坑协程调试与问题排查4.1 忘写await最常见的低级错误协程函数调用返回的是协程对象不是结果。忘写await时程序不会报错但结果完全不对。tasks [fetch_price(client, i) for i in range(100)] prices await asyncio.gather(*tasks)如果把fetch_price(client, i)误写成fetch_price(client, i)不加括号调用就会把协程对象传进列表然后gather等到的是一堆协程对象而不是价格数据。更隐蔽的是async def foo(): return 1 async def main(): result foo() # 忘写await print(result) # 打印 coroutine object foo at 0x...排查技巧如果你发现大量警告“coroutine was never awaited”说明有协程对象没被消费。养成立刻修复的习惯别让警告一直挂着。4.2 事件循环的重复运行问题在Jupyter Notebook或某些脚本环境里asyncio.run()只能调用一次因为run()会创建新的事件循环并在结束后关闭它。第二次调用会报错“Event loop is closed”。解决方案分几种脚本型程序标准写法就是asyncio.run(main())用完即走不要重复调交互式环境用asyncio.get_event_loop()和loop.run_until_complete(main())Python 3.12之后asyncio.run()在同一个线程里也不能重复使用了换环境重开就好。如果需要在已有的运行循环中嵌套跑协程要用asyncio.create_task挂进去而不是再调一个run_until_complete。4.3 同步阻塞代码混进协程这是协程性能杀手排行榜第一名。在协程里调用time.sleep()、requests.get()、pandas读取大文件、普通的open().read()都会阻塞整个事件循环。time.sleep()是最坑人的因为它看起来无害但一旦在协程里用了这个协程挂起后事件循环调度不到其他协程整个程序变成伪并发。正确做法是用await asyncio.sleep()——注意这个sleep是异步实现的会让出控制权。requests库同理要用httpx或aiohttp替代。数据库操作也要用异步驱动asyncpg是PostgreSQL的异步驱动aiomysql是MySQL的异步驱动redis-py自带redis.asyncio模块。如果你实在没办法替换某个同步库还有一种应急方案把它丢到独立线程池里去跑。import asyncio import requests async def fetch_sync_in_thread(product_id): loop asyncio.get_running_loop() # run_in_executor默认使用线程池执行器 result await loop.run_in_executor(None, requests.get, fhttps://api.example.com/price/{product_id}) return result.json()[price]这样虽然绕不开同步阻塞但至少不阻塞事件循环代价是每个调用占用一个线程池线程。线程池默认大小根据CPU核数决定注意别把所有任务都扔进去那样跟多线程没有本质区别。4.4 异常处理一个失败拖垮全部asyncio.gather默认是“一荣俱荣一损俱损”的任何一个任务抛出异常gather就会立刻向外抛异常其他任务的结果拿不到而且剩余任务可能被取消。async def main(): try: results await asyncio.gather(*tasks) except Exception: # 这里只会捕获到第一个异常 pass需要容忍单点失败时用return_exceptionsTrue参数results await asyncio.gather(*tasks, return_exceptionsTrue) for r in results: if isinstance(r, Exception): # 单独处理这个任务的异常 continue还有一个思路是用asyncio.wait它能更精细地控制等待策略比如FIRST_COMPLETED或FIRST_EXCEPTION适合做超时竞速、首响响应等高级调度场景。4.5 任务取消与超时控制IO任务经常需要设置超时防止某个请求卡死拖慢整体。asyncio.wait_for是标准做法import asyncio async def fetch_with_timeout(client, product_id): try: return await asyncio.wait_for( fetch_price(client, product_id), timeout3.0 ) except asyncio.TimeoutError: return None这里注意wait_for超时后会自动取消内部协程。如果你的协程里有finally块或者需要清理连接、回滚事务要捕获asyncio.CancelledError来做清理。注意Python 3.8以后CancelledError继承自BaseException不是Exception所以普通的except Exception捕不到它必须显式捕获。5. 协程生态与更多IO场景不止是HTTP请求5.1 异步文件操作对于文件IO普通读写是同步阻塞的。用aiofiles可以做到异步import asyncio import aiofiles async def read_file(path): async with aiofiles.open(path, r, encodingutf-8) as f: content await f.read() return content注意异步文件操作底层实际是走线程池实现的因为磁盘IO的异步原生支持在Python标准库里不完整。但它至少保证了你的事件循环不被文件读取卡死适合在异步流程中顺手做日志、读取配置文件等操作。5.2 异步数据库访问数据库连接在网络IO层面天然适合异步。以asyncpg为例import asyncio import asyncpg async def fetch_users(): conn await asyncpg.connect( userpostgres, passwordxxx, databasetest, host127.0.0.1 ) rows await conn.fetch(SELECT id, name FROM users WHERE status$1, active) await conn.close() return rows这里最关键的是数据库查询语句中用$1这样的序号占位符而不是%s或?。这是asyncpg的语法好处是类型推断更精确也避免SQL注入风险。初次切换asyncpg时最容易在这个细节上踩坑。MySQL方向用aiomysql接口风格与PyMySQL类似但走异步驱动。Redis方向用redis.asyncio操作风格与redis-py一致只把连接池初始化方式改为异步import redis.asyncio as aioredis async def redis_demo(): r aioredis.from_url(redis://127.0.0.1:6379/0) await r.set(key, value) val await r.get(key) print(val)5.3 协程与多进程结合IO密集CPU密集混合场景纯IO密集任务协程是首选。但如果任务里既有网络请求又有大量的本地计算比如解析大型JSON、图像处理、加解密单靠协程就不够了。因为协程切换是协作式的只要你进入一段没有await的计算逻辑整个事件循环就得等你算完。混合架构的通用模式是协程负责并发调度和IO等待CPU密集部分用ProcessPoolExecutor隔离到多进程。以一个爬虫加数据清洗的管道为例import asyncio from concurrent.futures import ProcessPoolExecutor async def process_row(row): loop asyncio.get_running_loop() # 把CPU密集的解析任务丢给进程池 return await loop.run_in_executor(process_pool, heavy_parse, row) async def main(): rows await fetch_all_rows() results await asyncio.gather(*[process_row(r) for r in rows])这种线程池和进程池的边界设计至关重要IO部分交给协程计算部分交给进程池。你只需要注意进程池跟事件循环的生命周期管理用async with ProcessPoolExecutor() as pool可以自动关闭。5.4 任务编排与结构化并发asyncio如果只用来做“并发发请求”确实大材小用了。它还提供了任务编排的原语让复杂的并发流程变得可控。举个例子你要同时抓取三个页面的数据三个页面抓完后做汇总统计但其中第二个页面必须在第一个完成后才能访问。这种情况用依赖编排async def page1(): return await httpx_get(https://api.example.com/page1) async def page2(dep): return await httpx_get(fhttps://api.example.com/page2?dep{dep}) async def page3(): return await httpx_get(https://api.example.com/page3) async def main(): # 阶段1: page1和page3并发 p1_task asyncio.create_task(page1()) p3_task asyncio.create_task(page3()) p1_result, p3_result await asyncio.gather(p1_task, p3_task) # 阶段2: page2依赖p1结果单独执行 p2_result await page2(p1_result) # 阶段3: 汇总 result {p1: p1_result, p2: p2_result, p3: p3_result} return resultasyncio还支持asyncio.Queue来做生产者和消费者模式适合爬虫中“页面提取链接-链接去重-下载内容”的多阶段流式处理。这种组合能力才是协程真正强大的地方它不只是加速而是重新定义了程序的组织方式。到此我的实际体会是协程不是一个“装上就能提速”的银弹它的效果取决于你是否用对了IO操作、是否合理控制了并发、是否处理好了异常和超时。如果这三样都拿捏住IO密集任务的卡顿问题基本可以宣告根治。真实生产环境里我几乎每个网络IO项目都会先按这个思路走一遍先把同步调用清理干净再用信号量限流最后补上超时和异常兜底三步操作做完程序就稳了。最后再分享一个小技巧调试协程代码时不要光看print输出用asyncio.all_tasks()打印当前事件循环的所有未完成任务能快速揪出哪些协程没退出、哪些永远挂起——在排查“程序结束但进程不退出”这类诡异问题时这招能省下你一下午的时间。