Python asyncio并发陷阱：我把一个IO密集型任务的执行时间从30分钟压到3分钟

Python asyncio并发陷阱我把一个IO密集型任务的执行时间从30分钟压到3分钟上周有个脚本跑了一个多小时还没结束我一度怀疑是机器出了问题。其实就是一个批量调用第三方API的定时任务从数据库里拉出一万条数据逐条调外部接口获取补充信息再写回数据库。逻辑简单到不能再简单但执行时间飙到了30分钟以上。看着终端里一条一条同步请求我突然意识到——这根本不是代码问题是并发模型选错了。问题同步请求的IO等待陷阱原始代码大概长这样importrequestsdefprocess_records(records):results[]forrecordinrecords:resprequests.get(fhttps://api.example.com/data/{record[id]})results.append(resp.json())returnresults表面上是个循环实际上每个requests.get()都在干等。网络往返假设500ms一万条数据就是 10000 × 0.5s 5000s差不多83分钟。实际因为API有速率限制和偶发超时跑了30多分钟已经算快的了。CPU占用率不到5%但任务就是跑不动。典型的CPU空转、IO阻塞场景。第一次尝试多线程但效果一般我第一反应是上多线程fromconcurrent.futuresimportThreadPoolExecutordefprocess_records_threaded(records,max_workers50):results[]withThreadPoolExecutor(max_workersmax_workers)asexecutor:futures[executor.submit(requests.get,fhttps://api.example.com/data/{r[id]})forrinrecords]forfutureinfutures:results.append(future.result().json())returnresults50个线程池跑下来时间压到了8分钟左右。 improvement 是有的但问题也很明显线程切换开销Python GIL 下50个线程的上下文切换不是免费午餐内存占用每个线程栈默认8MB50个就是400MBAPI 限流并发太高直接触发对方的429还得自己加退避逻辑能用但总觉得不够优雅。尤其是这个任务本质是IO等待用线程去抢CPU时间片有点杀鸡用牛刀。第二次尝试asyncio aiohttp质变我决定用 asyncio 重写。核心思路很简单一个线程内管理成千上万个协程IO阻塞时主动让出控制权网络响应回来再恢复执行。importasyncioimportaiohttpasyncdeffetch_one(session:aiohttp.ClientSession,record_id:str):urlfhttps://api.example.com/data/{record_id}asyncwithsession.get(url)asresp:returnawaitresp.json()asyncdefprocess_records_async(records,batch_size100):connectoraiohttp.TCPConnector(limit50,limit_per_host20)timeoutaiohttp.ClientTimeout(total30)asyncwithaiohttp.ClientSession(connectorconnector,timeouttimeout)assession:semaphoreasyncio.Semaphore(50)# 并发控制asyncdeffetch_with_limit(record):asyncwithsemaphore:returnawaitfetch_one(session,record[id])tasks[fetch_with_limit(r)forrinrecords]returnawaitasyncio.gather(*tasks)# 入口resultsasyncio.run(process_records_async(records))几个关键设计点1. TCPConnector 连接池limit50控制总连接数limit_per_host20避免对单一域名过度施压。实测下来这个参数不调的话aiohttp 默认会疯狂开连接对方服务器直接断开。2. Semaphore 并发限制50个并发是我根据对方API文档的限流策略100 req/s倒推的。Semaphore 保证了同时只有50个协程在执行网络请求其余的排队等待。3. gather 批量调度asyncio.gather()把所有任务一次性丢进事件循环由 loop 统一调度。不需要手动分片一万个协程在事件循环里跑内存占用只有协程对象本身的开销远小于线程。跑下来结果方案执行时间线程/协程数内存占用同步单线程~30分钟1~50MBThreadPoolExecutor(50)~8分钟50~450MBasyncio aiohttp(50)~3分钟1个线程 协程~80MB从30分钟压到3分钟内存还比多线程方案省了80%。踩坑记录坑1在 Jupyter 里直接await事件循环冲突Jupyter 本身已经运行了一个事件循环直接await process_records_async(records)会报RuntimeError: asyncio.run() cannot be called from a running event loop。解决办法importnest_asyncio nest_asyncio.apply()# 然后再 asyncio.run()或者直接用await而不套asyncio.run()。坑2aiohttp 默认不限制连接数导致对方拒绝服务第一次跑的时候没加TCPConnector(limit...)结果并发飙到几百API 直接返回 503。加 connector 限制后稳定了。坑3异常处理遗漏 gather 一个失败全崩默认gather()是一损俱损的某个请求抛异常整个结果列表拿不到。改成resultsawaitasyncio.gather(*tasks,return_exceptionsTrue)# 然后过滤掉异常实例success[rforrinresultsifnotisinstance(r,Exception)]坑4DNS 解析阻塞整个事件循环aiohttp 底层用 asyncio 的默认 DNS 解析某些环境下会同步调用系统 getaddrinfo卡住 loop。解决办法是装aiodnspipinstallaiodnsaiohttp 会自动识别并使用异步 DNS 解析器。写在最后这个案例让我重新思考了一个问题IO密集型任务到底该用什么并发模型CPU密集型计算、图像处理→ 多进程IO密集型 简单场景→ 多线程够用但不够精致IO密集型 高并发 资源敏感→ asyncio 是更优解Python 的 asyncio 不是银弹事件循环的调试和异常处理确实比同步代码心智负担重。但当你面对成百上千个网络请求需要并发调度时一个线程内跑几万个协程的能力是线程模型很难比拟的。最后贴一个我提炼的 checklist下次写批量IO任务之前翻一翻确认任务类型IO密集型还是CPU密集型评估并发上限对方API限流是多少我开多少合适异常兜底gather(return_exceptionsTrue)必须加连接池配置TCPConnector(limit...)必须设超时策略ClientTimeout防止请求挂死监控日志记录成功/失败/重试次数别闭着眼睛跑同步代码写了五分钟asyncio 重构加踩坑花了一下午。但看到这个脚本从30分钟压到3分钟的那一刻值了。