Python协程与asyncio核心概念详解：事件循环与异步IO实战入门

刚把多线程和多进程折腾明白的兄弟，估计又要被一堆概念绕晕了。协程、异步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 协程、线程、进程的对比清单

从表里能看出，协程的核心优势在于极低的切换开销和超高的并发能力，特别适合处理那种大量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里头可等待对象主要分三种：

• 协程对象（coroutine）
• Task对象（通过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开始引入asyncio，3.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 end

say_hello直到main完全睡醒之后才开始执行，整个程序退化成同步代码。因为time.sleep直接阻塞了底层线程，事件循环也被卡住了，根本没法切换去执行task。

记住这个原则：在asyncio代码里，所有同步阻塞IO的函数都必须换成对应的异步版本。time.sleep换成asyncio.sleep，requests换成aiohttp，普通文件读写换成aiofiles，数据库操作也尽量用异步驱动。否则你的协程代码只是披着异步的外衣，实际还是同步跑。

4. 用asyncio实现真正的并发：gather与create_task

4.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秒（最长的那个任务），而不是2+1+3=6秒。这就是并发带来的收益：三个请求同时发出，整体时间由最慢的那个决定。

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}, timeout=3) 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_asyncio

import 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.sleep
• requests→ 换成aiohttp或httpx
• 同步文件读写 → 换成aiofiles
• 同步数据库驱动 → 换成对应的异步版本

遵循原则：在一个协程函数里，凡是涉及IO取数据、发请求、等待时间的操作，都必须使用异步版本，否则协程并发就是纸上谈兵。

5.5 新手常见问题速查表

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)挂起，再去执行taskB，taskB也挂起，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的深层玩法：如何控制并发上限、异步队列、信号量、取消任务、以及处理超时——这些在实际项目里个个都是救命技能。