写 Python 写过一阵子的人多半都会碰上 asyncio 这个名字。我最初接触它是因为一次批量接口调用的需求两万个订单要逐个核对状态同步脚本跑一轮要一个多小时换成协程之后压缩到几分钟当时真有种原来还能这么写的感觉。这篇文章我就按自己踩坑的顺序把 asyncio 的事件循环、协程、任务几个核心概念拆开讲透再给出一套可以直接改成自己用的并发请求代码最后整理一些文档里查不到的问题排查经验。适合写过 Python 基础代码、想给脚本提速或者正准备用协程重写同步逻辑的朋友参考。1. asyncio 到底在解决什么问题1.1 从一个等接口的痛点说起很多人在学 asyncio 之前已经用 requests 写了不少同步脚本。同步脚本的逻辑很直白发一个请求等它返回再做下一件事。问题在于等这个动作太奢侈了。一次 HTTP 请求在网络往返上花 200 毫秒CPU 在这 200 毫秒里几乎什么都没干就是在空等。如果有一万次请求光等待时间就累加出两千秒但 CPU 的实际负载可能连 5% 都不到。当时我的第一反应是上线程池。线程池确实能提速但线程不是免费的每个线程有独立的栈空间线程切换要保存和恢复上下文线程数量一多系统的调度开销会明显上升。而且 Python 的 GIL 决定了线程在 CPU 计算阶段并不能真正并行对于 IO 等待场景线程的好处主要在于等待时不占用 GIL本质还是靠系统去切换。相比之下asyncio 的思路是在用户态自己管理切换单线程内部通过协作式调度把等待时间让给其他任务。这种方式的切换成本比线程低得多所以并发几千个任务也不会觉得吃力。需要先泼一盆冷水asyncio 只对 IO 密集型任务有效。所谓 IO 密集就是程序大部分时间花在等待外部资源上比如等数据库返回、等接口响应、等文件读写、等用户输入。如果你的任务是大量计算、循环、数据处理CPU 本来就满负荷协程的协作式切换不会带来任何收益那种场景应该考虑 multiprocessing 多进程。判断标准很简单看看程序在跑的时候 CPU 占用率是高是低低就是典型的 IO 密集。1.2 事件循环到底在忙什么理解 asyncio 最核心的一件事就是把事件循环这个词拆开看。事件循环是 asyncio 的中枢调度器它的行为可以概括成两步无限循环第一步检查当前有没有到期的定时器、就绪的 IO 事件、可以从挂起状态恢复的协程任务第二步把控制权交给其中一个任务让它一直执行到遇到 await 并让出控制权然后回到第一步继续挑选下一个任务。我用点外卖来类比。同步写法相当于一个人同时点了三份外卖非要等第一份送到才开始点第二份第二份到了再点第三份。线程池写法相当于雇三个人每个人盯一份外卖人力成本高人多了还互相拥挤。事件循环则是一个经验丰富的调度员他把三份订单都下出去商家备餐的间隙调度员转头处理别的事务哪一份外卖送到门口了他再去接哪一份。协程之所以能边等边做本质是把原本白白浪费的等待时间利用了起来。这里要澄清一个最常见的误解asyncio 不会让两个任务真正同时执行。单个事件循环在任意时刻只会运行一个协程协程之间是协作式切换不是抢占式。也就是说一个协程如果不主动 await 让出控制权事件循环就永远轮不到别的协程。真正的并行需要多个线程或者多个进程配合asyncio 从来不是平行的替代品。之所以这个模型在 IO 场景里这么高效是因为等待阶段 CPU 本来就没有计算可做让出控制权几乎零成本还能顺便跑完别的任务。2. 环境准备Python 版本与安装要点2.1 版本选择背后的原因动手之前先确认手上的 Python 版本。asyncio 对版本敏感程度比普通标准库高得多它从 3.4 进入标准库之后经历了三次大的形态变化早期只能用 asyncio.coroutine 装饰器和 yield from 写法别扭且难读3.5 正式引入 async/await 语法写法才变得正常3.7 加入 asyncio.run()事件循环的启动和关闭终于有了标准入口3.11 把 asyncio.TaskGroup 和 asyncio.timeout 这两个重磅特性正式落定异常安全性和超时控制都得到了实质性改善。我的建议是新项目直接用 Python 3.11 或者更高版本。原因不只是新 API 顺手还因为 3.10 前后事件循环的调度器实现做了不少优化同样的并发代码在较新版本上调度更顺滑。如果因为历史项目只能停留在 3.7 或 3.8也不至于没法用老 API 照样能完成绝大多数任务只是有些新写法要绕一下。下面所有代码都基于 3.11如果版本低个别语法需要替换我会在对应位置提醒。2.2 安装和第一行验证代码从官网 python.org 下载安装包是最直接的路径。Windows 用户记得在安装向导第一个页面勾选Add Python to PATH这个选项很多新手装完发现命令行敲 python 没反应九成是这一步没勾。macOS 用户可以用 Homebrew 安装Linux 发行版多半自带或者能用系统包管理器装。装完打开终端输入python -V看到 3.11 或更高的版本号说明基础环境就绪。Windows 下遇到 python 命令不识别的情况重装一遍并勾选 PATH 选项或者在系统设置里手动加上安装目录路径。接着写一个最小验证脚本确认 asyncio 能正常运行import asyncio async def main(): print(hello) await asyncio.sleep(1) print(asyncio works) asyncio.run(main())运行后先打印 hello等一秒再打印 asyncio works。这个脚本虽然简单但覆盖了 asyncio 最核心的三要素async 定义的协程函数、await 挂起点、asyncio.run 入口。如果输出符合预期你的环境就完全合格了。顺便提一句Python 3.4 以下没法用 async/await 语法真碰到这种老环境别折腾直接升级省心得多。3. 核心概念与常用 API 拆解3.1 async def 和 await 的真正含义一个函数前面加了 async调用的时候并不会执行函数体而是返回一个协程对象。这个对象只有进入事件循环才会真正运行。很多人第一次写 asyncio 都会在这里愣一下为什么我调用了函数它却什么都没打印因为协程对象的执行是惰性的。async def fetch_one(url): print(start, url) return ok print(fetch_one(http://example.com)) # coroutine object fetch_one at 0x...上面 print 输出的是一段协程对象的地址函数体里那一行 print 根本没执行。新手踩的第一个坑就是忘记 await导致协程对象创建了但从未运行程序不报错看起来像没反应。只要记住一条规则协程对象必须被 await、交给 create_task或者被 gather 这类高层 API 接管它才会真正开始执行。await 后面能跟什么能跟协程对象、Future 对象、Task 对象以及实现了await方法的对象。日常写代码遇到最多的就是协程和 Task。可以把 await 理解成挂起当前协程等目标完成后继续的语法标记。执行到 await 这一行时当前协程会把控制权交还给事件循环事件循环转而去跑其他就绪的协程等被等待的对象完成再回来接着往下走。3.2 create_task、gather 与 TaskGroup 怎么选并发执行多个协程最常用的有三个 API选哪个取决于你的具体需求asyncio.create_task(coro) 把协程包装成 Task 并立刻排进事件循环返回 Task 对象。你可以随时 await 它拿结果也可以单独 cancel 它。asyncio.gather(*coros) 一次性接收多个协程等全部完成之后返回结果列表顺序和传入顺序一致。适合等全部跑完再统一汇总的场景。asyncio.TaskGroup3.11用 with 语句圈定一组任务组内任何一个任务抛出异常会自动取消组内其他任务。它比 gather 更严格也更安全。需求推荐写法原因并发发请求最后统一收集结果gather return_exceptions代码短结果顺序与任务顺序一致需要中途单独取消某个任务create_task 保存 Task 引用Task 对象可以随时 cancel一个任务失败立刻取消其他任务TaskGroup组内任务取消有自动保证想每完成一个就处理一个as_completed不等全部完成逐个返回结果gather 有一个容易踩的坑如果其中一个协程抛出异常gather 默认会把这个异常直接抛出来导致后面的代码拿不到已完成的结果。如果你希望即使某个任务失败也能拿到其他成功结果一定要传 return_exceptionsTrue。这样异常会以对象形式待在结果列表的对应位置上而不是把整个 gather 打断。大批量请求场景里我几乎总是开着这个参数配合后续的状态码判断比让整个批次崩溃强得多。3.3 asyncio.run 的生命周期asyncio.run(coro) 是 3.7 之后最标准的入口函数。它内部做了三件事新建一个事件循环运行协程直到完成关闭事件循环并清理所有残留任务。这意味着你不应该在同一个程序里反复调用 asyncio.run也不该把 asyncio.run 内部使用的循环拿到外部去操作。正确用法是让 asyncio.run 包住你的顶层入口。import asyncio async def work(): await asyncio.sleep(0.1) return 42 result asyncio.run(work()) print(result)3.7 以前的老代码习惯手动 get_event_loop 再 run_until_complete现在除非你在写底层库或者需要深度控制循环否则 asyncio.run 就足够了。还有一个高频问题为什么在 Jupyter Notebook 里调用 asyncio.run 会报错因为 Jupyter 内核本身已经运行着一个事件循环你再用 asyncio.run 新建一个就会冲突。解决办法是把主逻辑包成 async 函数然后直接 await或者把脚本单独抽出来用命令行执行。这个问题在第五章会展开讲。4. 实操带限流和重试的批量并发请求4.1 方案取舍asyncio 最经典的实战场景就是批量 HTTP 请求。同步写法的痛点是请求耗时会全部累加而协程并发之后总耗时约等于最慢那一个请求的耗时脏活和累活都交给了网络等待的间隙。第三方库我推荐两个如果项目里已经习惯 requests 的 API 风格可以用 httpx 的异步客户端上手成本很低也可以用 aiohttp同样是老牌选择跟 asyncio 整合得很直接。下面代码用 aiohttp 演示。设计目标在写代码之前就定型并发请求 50 个假接口地址并发上限 10单次请求超时 5 秒失败自动重试一次最后统计成功率和总耗时。为什么并发上限要刻意限定因为无限并发会带来两个问题本机文件描述符会有上限一旦超出连接直接失败目标服务端也会因为瞬时打满连接而拒绝服务。限流不是锦上添花是批量请求的基本素养。4.2 完整可运行代码import asyncio import time import aiohttp SEM_LIMIT 10 # 并发上限 TIMEOUT 5 # 单次请求超时秒数 RETRY_TIMES 1 # 失败重试次数 TOTAL 50 # 请求总量 sem asyncio.Semaphore(SEM_LIMIT) async def fetch_one(session, index): url fhttp://example.com/api/items/{index} async with sem: for attempt in range(RETRY_TIMES 1): try: async with session.get(url, timeoutTIMEOUT) as resp: data await resp.text() return index, resp.status, data except asyncio.TimeoutError: print(f请求 {index} 超时第 {attempt 1} 次尝试) except aiohttp.ClientError as exc: print(f请求 {index} 失败{exc}第 {attempt 1} 次尝试) if attempt RETRY_TIMES: await asyncio.sleep(0.5 * (attempt 1)) return index, -1, async def main(): async with aiohttp.ClientSession() as session: tasks [asyncio.create_task(fetch_one(session, i)) for i in range(1, TOTAL 1)] results await asyncio.gather(*tasks) success [r for r in results if r[1] 200] print(f成功 {len(success)} / {len(results)}) for index, status, _ in results: print(fitem {index}: {status}) if __name__ __main__: start time.perf_counter() asyncio.run(main()) print(f总耗时 {time.perf_counter() - start:.2f} 秒)这段代码有几个细节值得展开。Semaphore 的 async with 包裹保证了任何时刻最多只有 SEM_LIMIT 个协程同时进入网络请求阶段超过的在信号量这里排队。重试逻辑写在 for 循环里超时和 ClientError 都进入重试分支每次失败后 sleep 一小段时间做退避避免重试风暴。把 index 放进返回元组里是为了 gather 完成后还能按原始编号对齐每个请求的结果。最后用 perf_counter 而不是 time.time 来统计耗时因为 perf_counter 基于高精度系统时钟不受系统时间跳变影响。4.3 核心参数调整建议并发上限不是越大越好。TCP 连接数、目标服务端处理能力、本机内存都是瓶颈。我的习惯是从 5 到 10 开始试先观察单次请求耗时和失败率再逐步往上调。如果单个请求本身只要 50 毫秒500 个请求用并发 20 基本能压到两三秒继续提高并发收益会明显递减失败率反而可能抬头。做压测时记录一组并发 5、10、20、50 分别的总耗时和成功数选一个拐点值这就是你的服务最合适的并发参数。超时参数尤其要留意。aiohttp 的 timeout 可以传秒数也可以传 aiohttp.ClientTimeout 对象分别控制连接建立、读取数据等阶段。在大批量场景里不设超时是灾难级的错误一个挂起的连接会一直占用信号量名额把整个批次拖成慢性死亡。重试也不是越多次越好对瞬时抖动来说一两次足够对持续故障来说重试再多也只是浪费时间。5. 常见问题与排查技巧实录5.1 RuntimeError: asyncio.run() cannot be called from a running event loop这个报错几乎总是出现在 Jupyter Notebook、某些自动化脚本框架这类已经内置事件循环的环境里。你写的代码本身没问题但 asyncio.run 要求从无循环状态启动一个干净的事件循环一旦宿主环境已经存在循环它就直接拒绝。解决办法有三个层次最推荐的是把主逻辑全部包进 async 函数在环境允许的地方直接 await 入口函数也可以用 nest_asyncio 给现有循环打补丁但那是临时方案代码里带着补丁总归不干净最稳妥的是把独立脚本抽出来用命令行运行彻底绕开宿主环境。5.2 协程里混入同步阻塞调用这是 async 程序最隐蔽的性能杀手。语法完全正确逻辑完全正确但执行时间莫名其妙地变成几倍几十倍。问题往往出在协程里偷偷调用了 time.sleep()、requests.get()、大文件整块读取这类同步操作。这些调用会像一颗钉子把事件循环整个钉住其他所有协程都暂停并发瞬间退化成串行。排查方法是观察现象50 个请求预计几秒跑完实际却跑了七八十秒第一个怀疑对象就是同步阻塞。把同步阻塞改成异步有三种常见的做法。首选是换异步库time.sleep 换成 asyncio.sleeprequests 换成 httpx 异步客户端或者 aiohttp。其次如果某个同步库没有异步版本又没法换可以用 loop.run_in_executor 把它丢到线程池执行让事件循环不至于被卡住import asyncio def blocking_work(x): import time time.sleep(1) return x * 2 async def main(): loop asyncio.get_running_loop() result await loop.run_in_executor(None, blocking_work, 21) print(result) asyncio.run(main())这里传入 None 表示使用默认线程池。要注意 run_in_executor 返回的 Future 可以被 await因此它不会阻塞事件循环。第三种办法是用多进程但那只适用于计算密集场景和 asyncio 属于不同维度别混为一谈。5.3 超时和取消的边界问题用 asyncio.wait_for 给任务加超时是常见需求但很多人没意识到超时触发时协程会被 CancelledError 取消。如果被取消的协程里正在使用外部资源比如已经建立了一个数据库连接或者网络句柄而协程内部没有善后逻辑资源就会泄漏。凡是可能被外部取消的协程资源获取和释放都应该用 try/finally 或者 async with 包住确保取消也能走完清理分支。另一个容易混淆的工具是 asyncio.shield。它的作用是保护一个协程不被外层取消。举个例子后台正在把一条重要记录写入数据库前端的请求超时了如果没有 shield超时取消会一路传进去把写库操作也打断。用 shield 包住写库部分之后即使外层超时内部的写操作也会继续完成。不过 shield 不是无限保护如果被 shield 的协程自身死循环整个事件循环仍然会被拖住。5.4 开启调试模式asyncio 自带了调试模式很多人没用过。用 asyncio.run(main(), debugTrue) 启动或者设置环境变量 PYTHONASYNCIODEBUG1。开启之后事件循环会记录两类重要信息一类是协程对象创建后从未被 await 的警告另一类是单个任务执行耗时过长的提示日志里会出现类似 Executing took X seconds 的文本。批量请求脚本性能不对劲时先开调试模式看看是不是某个慢任务把循环拖住了会比瞎猜快得多。6. 进阶思路标准库也能写并发请求6.1 基于 open_connection 的最小 HTTP 实现不引入任何第三方库asyncio 自带的 open_connection 也可以完成基础的 HTTP 请求。这节内容不是为了让你在生产环境手写 HTTP 协议而是为了让你看清网络库背后的原理HTTP 本质就是基于 TCP 的文本协议发送一段格式正确的请求文本接收服务端返回的响应文本。理解了这一点你对 aiohttp 之类的库就不会再有黑盒恐惧。import asyncio async def http_get(host, port80, path/): reader, writer await asyncio.open_connection(host, port) request ( fGET {path} HTTP/1.1\r\n fHost: {host}\r\n Connection: close\r\n \r\n ) writer.write(request.encode(utf-8)) await writer.drain() lines [] while True: line await reader.readline() if not line: break lines.append(line.decode(utf-8, errorsreplace).rstrip()) writer.close() await writer.wait_closed() return lines async def main(): lines await http_get(example.com, 80, /) for line in lines[:6]: print(line) asyncio.run(main())代码里值得注意的地方Connection: close 告诉服务端响应完就断开这样程序读取到 EOF 时就知道响应结束不用自己处理内容长度或者 chunked 编码writer.drain() 用于等待底层缓冲区真正把数据发送出去close 之后还要 wait_closed 确保连接完全关闭。生产环境仍然应该用封装好的库因为压缩、重定向、连接池、SSL 这些细节都已经被处理好了但这份最小实现能让你对整个过程有直观印象。6.2 用 asyncio.Queue 组织生产消费模型当任务流程分成多个阶段时队列是非常好用的缓冲工具。比如第一阶段从分页接口抓列表把抓到的条目放进队列第二阶段并发地从队列取条目逐条抓详情。asyncio.Queue 的用法和标准库 queue.Queue 很接近区别在于 get 和 put 都需要 await因为要保证并发安全并避免消费协程空转。import asyncio async def producer(queue): for i in range(10): await queue.put(i) await asyncio.sleep(0.01) async def consumer(queue): while True: item await queue.get() if item is None: break print(处理, item) async def main(): q asyncio.Queue(maxsize5) pro asyncio.create_task(producer(q)) cons asyncio.create_task(consumer(q)) await pro await q.put(None) await cons asyncio.run(main())这里我用 None 作为哨兵值表示生产结束。生成者协程完成后手动塞一个 None 进队列消费者收到就退出。特别提醒如果消费者有多个必须给每个消费者各发一个哨兵否则某些消费者会永远阻塞在 get 上。这是生产者消费者模型里最容易出 bug 的地方我在多人协作项目里已经见过好几次。6.3 代码骨架与排查顺序的心得把第 4 章的完整代码稍微扩展加上统一的失败日志输出和结果统计就可以作为一个通用的批量 IO 任务项目骨架复用。它覆盖了限流、重试、超时、结果汇总四个关键点这也正好是我认为 asyncio 实战中最重要的一组能力。当你想快速套用新场景时只需要把 fetch_one 里的请求逻辑替换掉其余结构可以直接保留。排查 asyncio 问题的时候我个人的定位顺序是固定的先确认 Python 版本是否满足所用 API 的最低要求再看事件循环层有没有抛出 RuntimeError然后用调试模式找慢任务最后检查协程里是否混入了同步阻塞。这个顺序能解决我遇到过的大多数问题从语法层面到性能层面基本都能覆盖到。我在实际使用中的一个体会是asyncio 的学习曲线不在语法而在思维方式的转变。写同步代码时你默认每一行都是立刻完成、顺序执行的写协程代码时你得时刻记得这里可能会等待等待期间别的任务在跑。只要跨过这道坎后面大多是熟悉 API 而已。如果你想动手试试建议直接从一个小批量请求脚本开始亲眼看到耗时从分钟级变成秒级的那一瞬间你就明白这套东西值不值得学了。最后一个小建议不要把并发数拉得过满留一些余量给目标服务也留一些余量给自己排查问题的空间这样跑批任务的时候心里才踏实。