Python asyncio 异步IO踩坑：我为什么把 100 个线程改成了 10 个 Event Loop

Python asyncio 异步IO踩坑我为什么把 100 个线程改成了 10 个 Event Loop说实话我一直对 asyncio 又爱又恨。去年接了一个高并发采集项目每天要拉取 50 万条外部 API 数据。一开始我信心满满Python 嘛上 ThreadPoolExecutor 就完事了100 个线程干它结果上线第一天服务器就给我上了一课。100 个线程的噩梦一开始的代码长这样fromconcurrent.futuresimportThreadPoolExecutorimportrequestsdeffetch(url):returnrequests.get(url,timeout30).json()withThreadPoolExecutor(max_workers100)asexecutor:resultslist(executor.map(fetch,urls))看起来没毛病对吧但跑了半小时我就发现不对劲CPU 没占多少内存却飙到了 8GBps -eLf | grep python | wc -l一看好家伙100 多个线程每个线程的栈内存 8MB光是线程栈就吃了 800MB更糟的是 GIL 锁线程多了反而互相抢锁API 响应时间从 200ms 涨到了 800ms我当时就懵了。这不是我想象中的高并发。换成 asyncio坑才刚刚开始行线程不行那就上 asyncio。听说这玩意儿号称能跑几十万并发我一想10 个 Event Loop 总够了吧结果第一个坑就让我踩实了importasyncioimportrequests# 这里埋雷了asyncdeffetch(url):returnrequests.get(url,timeout30).json()# 阻塞asyncdefmain():tasks[fetch(url)forurlinurls]returnawaitasyncio.gather(*tasks)asyncio.run(main())跑起来一看并发数确实上去了但 CPU 占用还是高。用asyncio.all_tasks()一查发现任务根本没并行执行而是一个接一个排队。requests 是同步库在 async 函数里调用会阻塞整个 Event Loop。这是我踩的第一个坑混用同步和异步代码。换上 aiohttp第二个坑来了好换成 aiohttpimportaiohttpasyncdeffetch(session,url):asyncwithsession.get(url,timeout30)asresp:returnawaitresp.json()asyncdefmain():asyncwithaiohttp.ClientSession()assession:tasks[fetch(session,url)forurlinurls]returnawaitasyncio.gather(*tasks)这下总算能并发了。但新的问题又来了aiohttp 的默认连接池太小。同时发 1000 个请求大部分时间都在等连接实际并发数只有几十个。查了半天文档发现要手动调limit参数connectoraiohttp.TCPConnector(limit200,limit_per_host50)asyncwithaiohttp.ClientSession(connectorconnector)assession:# ...调完这个参数吞吐量直接翻了 3 倍。10 个 Event Loop 的架构单个 Event Loop 再强也受限于单核。我用uvloop替换默认的 asyncio 事件循环再配合进程池搞了个 10 进程 每进程 1 个 Event Loop 的架构importasyncioimportuvloopfrommultiprocessingimportPool asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())asyncdefworker(urls_chunk):connectoraiohttp.TCPConnector(limit200,limit_per_host50)asyncwithaiohttp.ClientSession(connectorconnector)assession:tasks[fetch(session,url)forurlinurls_chunk]returnawaitasyncio.gather(*tasks,return_exceptionsTrue)defrun_worker(urls_chunk):returnasyncio.run(worker(urls_chunk))# 把 50 万 URL 分成 10 份每份 5 万chunks[urls[i::10]foriinrange(10)]withPool(10)aspool:resultspool.map(run_worker,chunks)内存从 8GB 降到了 1.2GB采集时间从 6 小时缩到了 45 分钟。那些没人告诉你的细节1. 异常处理必须加return_exceptionsTrueresultsawaitasyncio.gather(*tasks,return_exceptionsTrue)不然一个任务抛异常剩下所有任务都会被取消。我因为这个丢了一整批数据。2. 超时控制要分层aiohttp 的 timeout 不只是总超时要拆成连接超时和读取超时timeoutaiohttp.ClientTimeout(total30,# 总超时connect5,# 连接建立超时sock_read10# 读取超时)3. 别忘了关闭 Sessionasyncwithaiohttp.ClientSession()assession:# ...# 这里会自动关闭但如果是全局 session记得程序退出时手动 close我有一次把 session 定义成了全局变量程序跑完连接池没释放服务器端口被占满了。4. DNS 解析也是阻塞的默认情况下aiohttp 用系统 DNS 解析这个操作是阻塞的。高并发下建议装aiodnspipinstallaiodns然后在 ClientSession 里启用sessionaiohttp.ClientSession(connectoraiohttp.TCPConnector(use_dns_cacheTrue),trust_envTrue)写在最后从 100 个线程到 10 个 Event Loop这个改造让我明白了一件事高并发不是堆资源而是选对工具 调对参数。asyncio 确实能跑几十万并发但前提是你要知道它的脾气别混用同步代码连接池要手动调异常处理要到位DNS 解析别忽略如果你也在用 asyncio 做采集或者 API 聚合建议先检查一下这几处。很可能性能瓶颈不在 Python而在一个没注意到的默认配置。附一键诊断脚本importasyncioimportpsutilasyncdefdiagnose():procpsutil.Process()print(fCPU:{proc.cpu_percent()}%)print(fMemory:{proc.memory_info().rss/1024/1024:.1f}MB)print(fThreads:{proc.num_threads()})print(fConnections: len(proc.connections()))loopasyncio.get_event_loop()print(fEvent Loop:{type(loop).__name__})print(fPending tasks:{len(asyncio.all_tasks())})asyncio.run(diagnose())跑一下这个脚本看看你当前的 asyncio 程序是不是真的在并发还是在假装并发。