刚把多线程和多进程折腾明白的兄弟估计又要被一堆概念绕晕了。协程、异步IO、asyncio这些词听着高端其实解决的就是一个“程序跑得飞快但CPU却在空等”的问题。这篇是Python协程系列的第一篇我们先把异步IO和asyncio的核心概念彻底聊透再把事件循环、协程函数、Task这些家伙一个个拉出来遛一遛。内容主要面向已经掌握Python基础语法、准备往进阶走的同学也适合写过爬虫、被阻塞IO卡到怀疑人生的朋友——学完这一篇你就能用asyncio写出第一个真正异步并发的程序为后面深入异步编程打好地基。1. 先搞清楚协程到底要解决什么问题1.1 同步阻塞的痛爬虫老司机都懂传统同步代码的问题是效率太低先来看个最简单的例子。假设你去抓100个网页用requests这种同步库一个请求发出去之后代码就卡在那里等服务器响应这期间CPU明明啥事没有但你去干不了别的活。等服务器磨蹭了1秒才返回程序才继续往下走继续发第二个请求然后又等。100个请求串着来光等待时间就攒了100秒。有人会说了那我用多线程啊。线程本身并不轻量——每开一个线程操作系统要给它分配栈空间和内核资源线程的创建和切换也有开销。而且线程之间的竞争条件、锁问题写起来够你掉一层头发。关键还是那个“GIL”纯计算密集型的任务多线程在某些场景下不仅不加速反而会变慢。这就是异步IO要解决的痛点程序在等待网络响应、读写文件、查询数据库这种IO操作的时候不再傻傻阻塞住整个流程而是让出CPU去处理其他事情等IO完成了再回来接着干。1.2 异步IO其实就像快餐店排队这个概念说通了特别简单。你想象一家只有一个服务员的快餐店服务员接到订单之后并不会站在灶台前等菜做好而是把订单贴到墙上立刻回头接待下一位顾客。等后厨的菜出了服务员再端着菜找对应的顾客。这就是异步的思想发起请求之后不等结果先去干别的结果好了再回来取。对比一下同步模式的服务员接到一个订单就转进后厨盯着锅一道菜做完了才出来接下一个客人。单线程做菜当然慢多线程等于多雇几个服务员但你也得给每个人专门配后厨成本高。协程的做法就精巧多了——还是那一个服务员但他通过快速地切换注意力让所有顾客都感觉自己在被照顾。程序层面也一样event loop事件循环就是那个不停接单、派单、接收后厨通知的服务员协程就是那些等待完成的订单。1.3 协程、线程、进程的对比清单维度协程asyncio线程进程执行单位用户态轻量级调度内核态抢占式调度操作系统独立资源单位切换开销极小函数级切换中等涉及内核上下文切换大进程间通信成本高资源占用一个任务栈极小默认栈8MB左右独立地址空间最重适用场景IO密集型、高并发连接IO密集型、有CPU计算混合CPU密集型代码复杂度较低代码顺序书写高需处理锁高需处理IPC并发上限几万甚至几十万几百到几千视机器资源而定从表里能看出协程的核心优势在于极低的切换开销和超高的并发能力特别适合处理那种大量IO等待、但每条连接的计算量都不大的场景比如Web服务的高并发连接、大量网络请求、消息推送等等。当然如果是风景识别这种纯计算任务还是老老实实用多进程。1.4 什么时候用协程什么时候别用先说什么时候用。网络爬虫需要并发请求大量URL每个请求都是一次网络IO等待。Web服务端FastAPI、aiohttp这类高并发框架底层全是asyncio。消息队列消费者不断接收并处理消息每条消息都涉及磁盘或网络IO。实时数据采集需要同时监听多个数据源比如同时监听WebSocket和轮询HTTP接口。再说不适合的场景。纯CPU计算任务比如图像处理、复杂算法。协程切换并不能利用多核CPU因为整个进程还是被一个线程占用所以计算密集型的任务用协程不仅没有优势反而多一层调度开销。已经用了大量同步阻塞库的老项目改造成本高容易踩坑不如直接用多线程来得省事。2. asyncio模块的三大核心概念2.1 事件循环Event Loop事件循环是整个异步编程的心脏它在里面专门负责管理所有任务。说得直白点它就是一个死循环从任务队列里拿出一个协程执行当这个协程遇到IO等待比如await asyncio.sleep(1)时它会把控制权交回事件循环事件循环转头去执行下一个协程同时记录之前的协程在等什么当之前那个协程的等待条件满足比如IO完成、定时器到期事件循环就把它放回可执行队列继续往下跑。这个机制特别像快餐店服务员这个订单等菜就先服务下一个客人后厨喊菜好了再回来把这个订单送出去。Python 3.7开始官方推荐用asyncio.run()这个最顶层入口函数它会自动帮你创建事件循环、执行传入的协程、在结束时关闭事件循环。再也不用手动写loop asyncio.new_event_loop()这些代码了但这背后的机制还是要知道因为后面调试问题全靠它。2.2 async/await语法与原生协程在Python 3.5之前协程是通过asyncio.coroutine和yield from实现的写法和逻辑都比较绕。Python 3.5之后引入了async和await两个关键字原生协程的时代就到来了。async def定义的就是一个协程函数。注意调用这个函数并不会立即执行而是返回一个协程对象这个对象要被事件循环调度才会真正运行。打个比方就是你已经准备好了菜谱协程函数但还没有下单后厨还没开始做。await是协程的挂起点后面跟一个awaitable对象。当协程执行到await就会让出控制权等await的对象结算完成再回来继续。这个语法特别像在代码里打了个标记到这里我需要等待了先让别的任务跑一跑。注意await只能在async def定义的协程函数内部使用。在普通函数里写await解释器直接报SyntaxError没得商量。2.3 可等待对象与Task拿一个坑说新手最容易搞混的就是协程对象和Task的关系。前面说了async def函数调用后返回的是一个协程对象但它还没被规划为要执行的任务。Task就是事件循环用来调度和追踪协程的包装对象通过asyncio.create_task()来创建。那Task和协程对象有啥区别就好比你有一个项目计划协程对象和一位被安排了任务的员工Task。项目计划本身只是纸上的内容而被安排了任务的员工有明确的开始时间、进度追踪、甚至可以中途取消。asyncio里头可等待对象主要分三种协程对象coroutineTask对象通过asyncio.create_task()或loop.create_task()创建Future对象底层机制Task本身就是Future的子类一般用户不需要直接操作需要记住的关键点是如果你在写代码时收到提示RuntimeWarning: coroutine xxx was never awaited说明你创建了一个协程对象但忘了交给事件循环去执行。这是新手100%会遇到的坑后面讲。2.4 事件循环的工作机制从事件到回调事件循环里到底存了什么它维护了一个就绪任务队列和一个等待队列。当一个协程执行到await asyncio.sleep(1)时它在事件循环里注册了一个定时器回调然后这个协程被挂起。CPU空闲下来了吗没有事件循环马上从就绪队列里抓下一个协程执行。1秒后定时器到点事件循环把之前的协程重新放回就绪队列等它再次拿到执行权。所以并发数高不代表CPU忙只是事件循环在疯狂切换任务。event loop自身跑在一个单线程里这也是为什么协程能扛住几万并发——因为在那个线程里切换任务的开销只是一次函数调用的级别远低于操作系统的线程调度。3. 第一个asyncio程序从跑通到看懂3.1 环境准备确认Python版本写asyncio代码之前先确认你的Python版本。Python官方从3.4开始引入asyncio3.5加法了async/await语法3.7推出了asyncio.run()3.8之后事件循环策略更加完善。我建议最低使用Python 3.8以上版本最好直接用Python 3.10或更新的版本代码写起来省心官方文档里的示例也基本都是按新语法写的。检查版本很简单终端里跑这个命令python3 --version如果版本太低可以去官网下载安装新版本。Windows用户注意一下装完记得把Python加进环境变量PATH不然命令行里打python会找不到。3.2 第一个迷你异步程序直接上代码用一个最简单的案例体验一下协程的调度过程import asyncio async def say_hello(): print(开始执行 say_hello) await asyncio.sleep(1) print(say_hello 执行完毕) async def main(): print(main start) task asyncio.create_task(say_hello()) print(task 已创建但还没执行完) await asyncio.sleep(0.5) print(main 睡醒) await task print(main end) asyncio.run(main())运行结果长这样main start task 已创建但还没执行完 开始执行 say_hello main 睡醒 say_hello 执行完毕 main end注意输出顺序这就是异步调度的直观体现。main()里创建了task但say_hello不会立刻运行而是先打印“task 已创建”然后执行到await asyncio.sleep(0.5)这时main挂起事件循环把执行权交给say_hello所以打印了“开始执行 say_hello”。接着say_hello也遇到await asyncio.sleep(1)挂起。main的0.5秒先到期恢复执行打印“main 睡醒”。再等0.5秒后say_hello的1秒到期打印“say_hello 执行完毕”。最后main等task完成后结束。3.3 逐步拆解关键代码背后的设计逻辑先看asyncio.create_task()它会接收一个协程对象封装成Task并计划在事件循环下一次调度时执行。注意它是立即返回的不会等待协程执行完成所以task asyncio.create_task(say_hello())这行代码执行后程序并不会马上进入say_hello。再看await asyncio.sleep(0.5)这个函数和time.sleep()最大的区别在于time.sleep会阻塞整个线程期间事件循环也无法运行而asyncio.sleep会让出控制权让事件循环去干别的事。在await的瞬间当前协程就挂起了代码执行顺序被打破了。最后await task的意思是把task的结果等回来。如果task已经执行完成它立即返回如果还没完成当前协程会让出CPU等task结束后再恢复。3.4 别用time.sleep去混异步代码不夸张地说用time.sleep去替代asyncio.sleep是新手最容易犯的致命错误。试试把上面代码里的await asyncio.sleep(0.5)改成time.sleep(0.5)输出结果立马变成main start task 已创建但还没执行完 main 睡醒 开始执行 say_hello say_hello 执行完毕 main endsay_hello直到main完全睡醒之后才开始执行整个程序退化成同步代码。因为time.sleep直接阻塞了底层线程事件循环也被卡住了根本没法切换去执行task。记住这个原则在asyncio代码里所有同步阻塞IO的函数都必须换成对应的异步版本。time.sleep换成asyncio.sleeprequests换成aiohttp普通文件读写换成aiofiles数据库操作也尽量用异步驱动。否则你的协程代码只是披着异步的外衣实际还是同步跑。4. 用asyncio实现真正的并发gather与create_task4.1 先看并发和并行这俩词很多人把并发concurrency和并行parallelism混着用但在异步编程里这俩区别大了。并行是同时执行多个任务需要多个CPU核心比如多进程在4核CPU上同时跑4个进程。而并发是多个任务交替执行单核就可以实现就像事件循环在单线程里快速切换协程一样。对于IO密集型任务并发就够用了——因为等待时间才是大头真正在CPU上跑的代码很少。所以asyncio实现的这种“伪并行”大多数人觉得它本事大其实是因为网络等待时间特别长交替执行能极大提升吞吐量。这一点在爬虫场景尤其明显。4.2 看一段并行爬取示例简化版这里用一个模拟函数代替真实网络请求演示多任务并发import asyncio async def fetch_url(name, delay): print(f开始请求 {name}) await asyncio.sleep(delay) # 模拟网络IO等待 print(f请求完成 {name}耗时 {delay} 秒) return f{name} 的数据 async def main(): # 注意create_task之后这些协程就已经被调度开始并发执行 task1 asyncio.create_task(fetch_url(首页, 2)) task2 asyncio.create_task(fetch_url(列表页, 1)) task3 asyncio.create_task(fetch_url(详情页, 3)) # 依次等待任务完成 result1 await task1 result2 await task2 result3 await task3 print(result1) print(result2) print(result3) asyncio.run(main())运行结果开始请求 首页 开始请求 列表页 开始请求 详情页 请求完成 列表页耗时 1 秒 请求完成 首页耗时 2 秒 请求完成 详情页耗时 3 秒 列表页 的数据 首页 的数据 详情页 的数据注意“开始请求”三个全部秒出而不是等第一个2秒后才发第二个。整个程序总耗时约3秒最长的那个任务而不是2136秒。这就是并发带来的收益三个请求同时发出整体时间由最慢的那个决定。4.3asyncio.gather与asyncio.wait批量管理任务如果每次都要手动创建task再挨个await代码会有点啰嗦。asyncio提供了gather和wait两个工具函数来简化批量任务操作。asyncio.gather的场景是把多个协程并发执行并等待全部结束后拿到结果列表。它的优点是返回结果是按传入顺序排列的哪怕task完成有先有后。import asyncio async def fetch_url(name, delay): await asyncio.sleep(delay) return f{name} 的数据 async def main(): results await asyncio.gather( fetch_url(首页, 2), fetch_url(列表页, 1), fetch_url(详情页, 3), ) print(results) # 输出: [首页 的数据, 列表页 的数据, 详情页 的数据] asyncio.run(main())asyncio.wait的使用方式和gather略有不同它接收的是Task对象的集合返回值是两组Set完成的任务和未完成的任务。适合那种需要中途做判断、不需要等全部跑完的场景import asyncio async def say_after(delay, word): await asyncio.sleep(delay) return word async def main(): task1 asyncio.create_task(say_after(1, hello)) task2 asyncio.create_task(say_after(2, world)) done, pending await asyncio.wait({task1, task2}, timeout3) print(已完成:, done) print(未完成:, pending) asyncio.run(main())实操建议如果你只是想让多个协程并行跑然后汇总结果无脑用asyncio.gather就对了。如果你需要控制超时时间、中途取消某几个任务、或者按任务完成先后逐个处理结果那才需要asyncio.wait。4.4 为什么协程适合IO密集而不适合CPU密集前面已经提过很多次IO密集和CPU密集再说得透一点。IO密集任务的CPU占用率很低大部分时间都在等待。比如爬虫发一个HTTP请求CPU真正工作的时间可能只有几毫秒剩下的999毫秒都是在等服务器响应。asyncio只用一个线程就能在这个等待时间里来回切换执行成百上千个请求把等待时间压缩到极致。CPU密集任务恰恰相反比如对大数组排序、图像卷积计算CPU一直在忙。asyncio切换来切换去无非是把CPU时间片分配给不同协程但总计算量并没有变少单个线程也利用不了多核所以性能反而会因为上下文切换的开销而略有下降。如果一定要让asyncio处理CPU密集型任务一个常用的方案是用asyncio.to_thread()或者loop.run_in_executor()把计算子任务丢到线程池里去跑让事件循环不被阻塞。不过这属于进阶玩法这篇先不展开。5. 碰到过的坑asyncio常见问题排查实录5.1 最常见的警告coroutine was never awaited这个警告我敢打包票每个接触asyncio的人都遇到过一次。典型场景import asyncio async def hello(): print(hello) async def main(): hello() # 忘了写 await asyncio.run(main())运行后你会在控制台看到RuntimeWarning: coroutine hello was never awaited原因前面说过了hello()返回的是协程对象你如果不await它它就永远不会执行。实在忘了写await哪怕直接不调用这个协程函数解释器也会给你警告。写代码的时候养成好习惯调用协程函数时要么直接await要么用asyncio.create_task()包成Task要么传给asyncio.gather()。否则只要出现警告大概率代码逻辑就是有问题的。5.2 Jupyter Notebook里跑asyncio直接报错Jupyter里跑这段代码会报一个诡异的错误RuntimeError: This event loop is already running原因是Jupyter自身有事件循环在跑而asyncio.run()会尝试创建新的事件循环结果冲突了。解决方案是安装nest_asyncio库然后开头加上这几行pip install nest_asyncioimport nest_asyncio nest_asyncio.apply()这个库的作用是把当前运行中的事件循环嵌套一层允许新的循环进入。不过不建议在正式生产代码里用它只是为了解决开发调试环境的限制。生产环境老老实实把入口写成一个函数用asyncio.run()来启动。5.3 在同步代码里调用异步函数遇到的情况是有个人写了一个同步函数在里面直接调用了协程函数结果又碰到was never awaited的警告。很简单的解决办法加个asyncio.run()包住协程调用。import asyncio async def fetch(): await asyncio.sleep(1) return data def sync_function(): result asyncio.run(fetch()) print(result) sync_function()但要注意asyncio.run()不能在已经运行的事件循环里调用也就是说不能在一个async函数里再嵌套调用asyncio.run()。例如import asyncio async def inner(): return 1 async def outer(): result asyncio.run(inner()) # 报错这会在运行时抛出RuntimeError: asyncio.run() cannot be called from a running event loop。解决办法就一条async函数里统一用await不要在async里面开新的事件循环。5.4 协程里混入同步阻塞代码并发全废还有一个非常隐蔽的问题。你在某个协程函数里用了requests.get()或者time.sleep()这种同步阻塞调用看起来程序能跑但并发性能会断崖式下降。原因还是那个阻塞调用会阻塞住整个事件循环所在的那个线程。结果就是即使你创建了100个task只要其中一个task执行了time.sleep(1)这个1秒内事件循环就完全卡死其他task全部陪跑。所以排查并发问题的时候优先检查代码里有没有这些雷time.sleep→ 换成asyncio.sleeprequests→ 换成aiohttp或httpx同步文件读写 → 换成aiofiles同步数据库驱动 → 换成对应的异步版本遵循原则在一个协程函数里凡是涉及IO取数据、发请求、等待时间的操作都必须使用异步版本否则协程并发就是纸上谈兵。5.5 新手常见问题速查表问题症状解决方法协程未被awaitRuntimeWarning: coroutine was never awaited补上await或用create_task/gather包装在Jupyter里运行报错RuntimeError: This event loop is already running安装并调用nest_asyncio.apply()async函数里调用asyncio.runRuntimeError: asyncio.run() cannot be called from a running event loop用await替代asyncio.run混入time.sleep并发失效变成顺序执行换成asyncio.sleep忘了在async函数里写await协程不执行或警告检查代码逻辑确认所有await都正确用同步requests库并发时性能极差换成aiohttp等异步HTTP库5.6 排查asyncio问题的思路总结遇到asyncio相关的问题我的排查顺序一般是这样先看警告信息是RuntimeWarning还是RuntimeError前者多半是忘了await后者多半是事件循环冲突。然后打印日志在每个协程的开始和结束各加一行print观察具体执行顺序判断是不是有阻塞调用混进来了。快速实践一下用一个同步阻塞调用污染协程看看输出的顺序变化你就能直观体会到阻塞对事件循环的破坏力。这是理解asyncio调度机制最有效的实操方式。6. 多任务之间的协作方式与执行顺序6.1 await和create_task两种创建任务的姿势差异新手经常搞不清await asyncio.sleep(1)和task asyncio.create_task(fn())之间的执行差异再拿代码来说话import asyncio async def step1(): await asyncio.sleep(1) return 1 async def step2(): await asyncio.sleep(1) return 2 async def sequential(): start asyncio.get_event_loop().time() await step1() await step2() print(顺序执行耗时:, asyncio.get_event_loop().time() - start) async def concurrent(): start asyncio.get_event_loop().time() t1 asyncio.create_task(step1()) t2 asyncio.create_task(step2()) await t1 await t2 print(并发执行耗时:, asyncio.get_event_loop().time() - start) async def main(): await sequential() await concurrent() asyncio.run(main())顺序执行大约2秒并发执行大约1秒。原因在于await step1()会等step1完全结束才执行step2而create_task后两任务同时被事件循环调度它们的耗时重叠了。这就是为什么凡是需要并发执行的协程必须先通过create_task或gather来调度而不是直接在main里依次await两个协程。6.2 事件循环的任务调度机制深入理解事件循环本质就是一个优先级队列加一堆回调函数的组合。当协程A执行到await时它的状态被保存函数栈、局部变量控制权归还给事件循环。事件循环从就绪队列里取出协程B开始执行。B遇到await后同样挂起事件循环继续找下一个。这就像快餐店那个服务员他永远在不停接单、下单、返回顾客、返回厨师循环往复。关键是他同一时刻只处理一个顾客的对话但由于切换速度极快微秒级宏观上看就像同时服务了所有顾客。协程的切换完全由程序自己控制遇到await就挂起不像线程那样会被操作系统随时抢占。这种协作式调度让并发行为变得可预测也大幅降低了竞态条件出现的概率但代价是——如果某个协程故意不屈服比如写了一个死循环还不带await事件循环就永远没机会调度其他协程了。6.3 实现一个协程版的倒计时器巩固理解写代码永远是掌握概念的最好方式。下面这个倒计时器用协程实现了两个倒计时的交替执行import asyncio async def countdown(name, total): print(f{name}: 开始总时长 {total} 秒) for remaining in range(total, 0, -1): print(f{name}: 剩余 {remaining} 秒) await asyncio.sleep(1) print(f{name}: 结束) async def main(): task_a asyncio.create_task(countdown(A, 3)) task_b asyncio.create_task(countdown(B, 2)) await task_a await task_b print(全部倒计时结束) asyncio.run(main())运行后你会看到A和B的“剩余”消息交替输出而不是A倒计时完了再开始B。这就把协程并发调度的感觉整明白了。从“同步心法”角度理解这段代码事件循环把taskA拿出来执行到await asyncio.sleep(1)挂起再去执行taskBtaskB也挂起1秒后taskA先到期恢复执行打印第二次剩余继续挂起taskB恢复打印如此交替。整段代码只占一个线程却同时维护了两个“虚拟时钟”。6.4 注意协程之间的执行顺序并不确定一个需要提前打好预防针的点多个协程并发执行时它们进入await之后谁先恢复完全取决于事件循环里的回调顺序和定时器时间并不是按照创建Task的先后顺序来的。举例来说你在main里先创建了task1再创建task2但不代表task1一定先执行完。如果task1要等3秒task2只要等1秒那task2肯定先完成。所以千万别在代码逻辑里依赖并发任务的执行顺序——需要保证先后关系时老老实实串行await或者用asyncio.gather里传参顺序来保证结果顺序。7. 这一篇的收尾与下篇预告看到这里asyncio的最核心骨架已经搭起来了事件循环负责调度async def定义协程await挂起并让出控制权create_task创建任务实现并发gather批量管理任务。异步IO的核心思想就一句话——把等待时间干别的活。我在实际写async代码的时候有个体会别一上来就追求花哨的高并发技巧先把这几个基本的执行模型在脑子里练熟。每次写await都问自己一句“此刻事件循环有时间去跑别的任务吗”时间长了你的异步直觉自然就出来了。踩过几次坑之后我给新学asyncio的同学的建议是代码里严格遵守“async函数里只放异步操作”这个原则写之前先明确任务类型是IO密集还是CPU密集再把任务编排方式选好。这篇先把地基打牢下一篇我们聊asyncio的深层玩法如何控制并发上限、异步队列、信号量、取消任务、以及处理超时——这些在实际项目里个个都是救命技能。