Python异步编程入门：asyncio协程与事件循环实战指南

如果你用Python写过爬虫，大概率经历过这种尴尬：请求发出去之后，程序眼巴巴等着服务器返回，同一时刻CPU几乎闲置，可整个脚本的耗时却全被网络延迟拖垮。几千个URL要抓，每个请求两三百毫秒，同步写下来就得跑十几分钟，人守着命令行干着急。这种场景下，Python异步编程里的asyncio库就是最对症的解法——它能在单线程内反复调度任务，让程序在等待网络响应的时候，转手去处理其他请求。换句话说，异步不是让CPU“跑得更快”，而是让CPU“不再闲着”。

如果你是刚开始接触asyncio，这篇内容会从“为什么需要异步”讲起，把事件循环、协程、Task这些基础概念一个个掰开揉碎，再带你把代码跑起来，最后整理一些我在实际项目中踩过的坑。不需要你先懂并发理论，只要写过Python、知道函数怎么调用，就能跟上节奏。

1. 为什么要学异步：先算清“阻塞”这笔时间账

1.1 一次网络请求的“时间账本”

很多初学者学异步时，第一个疑问是：我的循环不是照样在跑吗，换成异步怎么就能快那么多？要回答这个问题，最好把一次普通HTTP请求的时间账算清楚。

假设你向某个接口发一个请求，它的响应时间是100毫秒。同步代码里，这一行requests.get(url)会发生什么？你的线程要把请求发出去，然后原地等待，直到100毫秒后响应数据抵达，代码才继续往下走。在这100毫秒内，这个线程除了盯着网络套接字，什么也干不了，CPU也处于“无事可做”的状态。如果你有一万个请求要发，每个请求都这么等待，总耗时就是一万乘以100毫秒，约等于16分40秒。

但如果你把“等待响应”这个时间段单独拿出来看，其实每个请求真正需要CPU参与的时间只有几毫秒——构造请求、解析响应、保存数据。剩下的95%以上的时间都花在等待上。同步脚本的问题不是“干活慢”，而是“等待多”：线程在等待时被牢牢占住，一个线程一次只能等一个请求，资源利用率低得惊人。

我常跟朋友打一个比方：同步代码像一个人在食堂打饭，点完A窗口的菜之后就一直站在窗口前面等，等到出餐了才离开。如果一次要打五六个菜，这个人就只能在各个窗口前轮流干等。异步代码则是点完A窗口之后先跑去B窗口点单，再跑去C窗口，等大家都快好了，再回头挨个把菜端走。窗口的出餐速度没变，但同一个人在同一段时间里能同时张罗的事情多得多。

1.2 并发不是为了“跑得快”，而是为了“不闲着”

这里要澄清一个很常见的误区：异步并发并不会让单个任务的执行速度变快，它提升的是整体吞吐量。所谓吞吐量，就是单位时间内能处理多少个任务。网络爬虫、批量接口调用、大量文件读写、数据库查询——这些场景的共同特点是“任务和任务之间互不依赖”，等待的间隙完全可以用来做别的事情。

只要等待的时间远大于处理的时间，异步就能带来肉眼可见的收益。反过来，如果你的任务是纯计算型的，比如跑一个大数乘法、做图像像素级的滤镜处理，这种场景CPU在持续工作时本来就没有“闲着”，asyncio不但不会提速，反而会因为任务调度的开销让性能更差。所以判断一个场景适不适合上asyncio，就看一个指标：你的程序里，CPU真正忙的时间占比高不高。不高，就非常适合异步。

2. 四个核心概念：协程、事件循环、Task与Future

2.1 async def与协程对象：调用它不等于执行它

初学者的第一个崩溃点，往往在这里：为什么我调用了一个async def定义的函数，里面什么都没打印？

import asyncio async def hello(): print("Hello") await asyncio.sleep(1) print("World") coro = hello() # 这里没有打印任何东西

用async def定义的函数，调用时不是执行函数体，而是创建一个协程对象。这个对象像一个“尚未开始的计算”，它等着被“调度”才会真正运行。要让协程里的代码执行，有两种方式：第一种，在另一个协程内部用await去等待它；第二种，把它包装成Task交给事件循环去调度。

async def main(): await hello() # 这时候才会打印 Hello / World asyncio.run(main())

await这个关键字可以理解为“在这里等待，同时把控制权交还给事件循环”。当协程执行到await的时候，它会检查被等待的那个东西是否已经完成，如果没有完成，就把自己“挂起”，让事件循环抽出时间去运行其他任务。等到被等待的东西准备好了，再回到这里继续往下执行。这个“挂起-唤醒”的机制，就是异步编程的核心动作。

2.2 事件循环：全局调度枢纽

你可以把事件循环想象成一个调度中心。所有的协程都注册到这里，事件循环手里维护着多个队列：哪个协程在等待网络I/O、哪个协程在等待定时器、哪个协程已经准备好被继续执行。它不断地循环问自己：现在有没有什么任务准备好了？有，就唤醒它，让它跑一小段；没有，就让出CPU，进入休眠等待下一次外部事件。

在Python 3.7之后，普通异步程序的入口非常简单：

asyncio.run(main())

这一行代码干的事情其实不少：它内部会创建一个新的事件循环，把main()这个协程丢进去执行，等main()彻底运行结束后关闭这个事件循环，完成清理工作。事件循环既不需要你手动创建，也不需要你手动关闭，asyncio.run()把这一切都自动化了。

2.3 Task与Future：谁负责让协程真正跑起来

单纯的协程对象还不足以被事件循环调度。要让协程真正“并发”起来，你得把它包装成Task。Task在事件循环中代表一个正在并发运行的子任务。看个例子：

async def main(): task1 = asyncio.create_task(hello()) task2 = asyncio.create_task(hello()) await task1 await task2

调用asyncio.create_task(hello())的那一刻，协程就已经被登记到事件循环里，即将开始运行。注意这里不需要再await hello()，而是await task1——Task对象在后台持续推进，你随时可以等它完成。

那Future又是什么？Future是一个更底层的概念，你可以把它理解成“一个未来的结果”。它本身不包含任何业务逻辑，只是一个占位对象，表示某项操作的结果“早晚会来”。Task继承了Future，相当于把“协程任务”和“结果占位符”两者合一：一次任务调度的结果，最终都会落到Task这个Future子类里面。

这三个概念的关系，可以归纳成一句话：协程是任务的内容，事件循环是任务的调度器，Task是任务的调度单位，Future是任务结果的容器。搞懂这四个词，asyncio的地基就算打牢了。

2.4 概念关系速查

3. 第一个异步程序与三种并发调度方式

3.1 用sleep体验并发：同样的两秒任务，从4秒变2秒

最直观的体验并发方式，是写两个 каждый睡2秒的任务，分别用同步和异步来跑。下面是完整的对比代码：

import asyncio import time async def worker(name, delay): print(f"任务 {name} 启动") await asyncio.sleep(delay) print(f"任务 {name} 完成") return f"{name} 的返回值" async def main(): # 同步串行的写法 start = time.perf_counter() await worker("A", 2) await worker("B", 2) print(f"串行耗时: {time.perf_counter() - start:.2f}s") # 异步并发的写法 start = time.perf_counter() results = await asyncio.gather( worker("C", 2), worker("D", 2), ) print(f"并发耗时: {time.perf_counter() - start:.2f}s") print(results) asyncio.run(main())

输出里你会看到两段非常直观的差异：串行版本打印“A启动”后等2秒，再打印“B启动”，总耗时4秒；并发版本中“C启动”和“D启动”几乎同时出现，2秒后两个任务一起完成，总耗时2秒。asyncio.sleep和time.sleep的区别就在这：time.sleep会把线程往死里阻塞，CPU空转；asyncio.sleep则是一个可等待对象，遇到它时协程会让出控制权，等时间到了再被唤醒。

3.2 create_task、gather、wait：三者怎么选

asyncio.create_task、asyncio.gather、asyncio.wait都可以实现多个任务的并发调度，但它们的设计目标有微妙差别，新手很容易搞混。

create_task适合“创建后台任务后，主流程还要继续做别的事”的场景。比如先启动三个下载任务，再去做数据清洗，最后统一等待结果。用create_task创建Task时，任务已经在运行了，不需要await它也能推进。但要注意：如果你创建了Task却从不等待它，程序退出时它可能被静默取消。一个稳妥习惯是给所有Task都留一个await点。

gather适合“等一批任务全部完成，然后收集结果”的场景。它接受多个协程或Task，按传入顺序返回对应的结果列表。gather还有个实用细节：如果其中某个任务抛异常，默认情况下gather会直接把异常抛出来，其他任务的结果即使成功你也拿不到。如果想保持“一个失败不拖垮整体”的行为，得把return_exceptions=True传进去。

wait比gather更底层，它接收的是Task集合，允许你通过return_when参数控制返回时机。asyncio.ALL_COMPLETED是所有任务完成后返回，asyncio.FIRST_COMPLETED是只要有一个任务完成就立刻返回。后者在“多个上游接口任一个返回即可继续”的场景很有用。wait返回一个元组(done, pending)，分别表示已完成和未完成的Task集合。

从实践经验来说，日常业务里90%的情况用gather就够了。等到你需要“先到先得”的响应策略，或者要对任务集合做更精细的生命周期管理时，再考虑用wait和create_task组合。

3.3 asyncio.run()的版本变迁与使用边界

asyncio.run()是Python 3.7引入的顶层入口函数，它帮你处理了事件循环的创建、运行、关闭这些杂活。到了Python 3.10，官方把asyncio.run()里的loop参数移除了，也就是说你不能再显式传入一个现成的事件循环。这在日常使用中反而是好事，省去了一堆“循环重用”的坑。

有几个边界值得记住：第一，同一个线程中不能嵌套调用asyncio.run()，否则会抛出RuntimeError: This event loop is already running；第二，asyncio.run()执行完后会关闭事件循环，所以不能指望它被反复调用去“重启”同一个loop；第三，在Jupyter Notebook这类本身已经维护着事件循环的环境里，直接调用asyncio.run()可能出问题，推荐用await main()代替。

4. 异步生态的进阶工具箱

4.1 async with与async for：不仅函数可以异步，上下文也可以

如果你只学会了async def和await，能覆盖的场景其实有限。真实项目里还有两个高频需求：异步资源的获取与释放，以及异步流的逐条处理。这就是async with和async for的用武之地。

异步上下文管理器对应普通上下文管理器，只不过它的__enter__和__exit__换成了异步版本__aenter__和__aexit__。写起来是这样：

class AsyncResource: async def __aenter__(self): print("获取资源") await asyncio.sleep(0.1) return self async def __aexit__(self, exc_type, exc, tb): print("释放资源") await asyncio.sleep(0.1) async def demo(): async with AsyncResource() as res: print("使用资源中")

最典型的使用场景就是异步HTTP会话。很多异步请求库要求你手动管理会话连接，用async with可以保证会话在使用完后被正确关闭，避免连接泄漏。

async for则用于异步迭代器。它背后的协议是__aiter__返回异步迭代器自身，__anext__返回一个可等待对象，迭代结束抛StopAsyncIteration。下面是一个简单的例子：

class AsyncCountdown: def __init__(self, n): self.n = n def __aiter__(self): return self async def __anext__(self): if self.n <= 0: raise StopAsyncIteration self.n -= 1 await asyncio.sleep(0.2) return self.n + 1 async def main(): async for num in AsyncCountdown(3): print(num)

实际项目里你很少会自己手写异步迭代器，更多是消费别人提供的异步流，比如逐行读取一个大文件、分批消费消息队列，都可以用async for写出非常干净的代码。

4.2 超时与取消：wait_for和cancel是异步程序的安全带

异步程序跑起飞之后，最怕的就是某个协程迟迟不返回——可能是下游接口挂了，可能是对方网络断了。如果你不设超时，整个程序会一直挂在那里。这就是asyncio.wait_for的用武之地：

async def main(): try: result = await asyncio.wait_for(worker("A", 5), timeout=2) except asyncio.TimeoutError: print("任务超时，已经被自动取消")

wait_for会在超时后自动取消内部任务。注意它抛出的超时异常是asyncio.TimeoutError，在Python 3.11之前它其实是concurrent.futures.TimeoutError的别名，捕获的时候写asyncio.TimeoutError最稳妥。

除了超时，你还可以手动取消任务：

task = asyncio.create_task(worker("B", 10)) await asyncio.sleep(1) task.cancel() # 请求取消 try: await task except asyncio.CancelledError: print("任务被取消了")

取消一个Task后，在await它的地方会抛出CancelledError。如果你的业务代码里有finally清理逻辑，它仍然会被执行，这给了你处理“被取消后释放资源”的机会。Python 3.11之后还推出了asyncio.timeout异步上下文管理器，语义上更清晰，允许你把超时和业务代码放得更贴合：

async def main(): try: async with asyncio.timeout(2): await worker("C", 5) except asyncio.TimeoutError: print("超时啦")

4.3 同步代码改异步：一个库替换清单

要真正享受异步的好处，光写async def没用，底层那些阻塞调用必须换成异步版本。最常见的替换是这三个：

time.sleep换成asyncio.sleep，这是最简单的。最难改的是用requests这种同步HTTP库。requests的每个请求都会阻塞线程，在异步协程里调用它，等于把整个事件循环都堵住，再多并发任务也白搭。解决办法是换成aiohttp或者httpx的异步客户端。httpx的API设计更接近requests，迁移成本相对低一些：

import httpx async def fetch(url): async with httpx.AsyncClient() as client: resp = await client.get(url) return resp.text

如果某个同步库实在没有异步替代品，还有一个退路：asyncio.to_thread。它可以把一个同步函数丢到单独的线程池里执行，避免阻塞事件循环：

import requests result = await asyncio.to_thread(requests.get, url)

注意to_thread本质上是“用多线程掩盖同步阻塞”，线程本身还是有开销的，适合偶尔调用，不适合一秒钟几百次高频使用。真正的解法永远是找到对应的异步库。

5. 实战排查：那些坑过无数人的问题

5.1 症状：异步里塞了同步阻塞库，整个程序卡住

这是我在实际项目里遇到最多的问题。有人会把requests放在异步函数里，然后跑gather，结果发现程序不但没变快，其他任务反而全都被拖住了。

原因其实不复杂：事件循环只有一个线程，当这个线程去执行requests.get时，它是在同步地等待网络响应，根本不会回到事件循环去调度其他协程。哪怕你对十个URL并发请求，结果还是十个请求排队等待。

排查方式也很直接：看看协程里调用的库是不是支持异步。如果某个库的文档里没有“async”字样，它的调用基本都会阻塞线程。判断不了的时候，可以在代码里临时加打印，看任务是不是按顺序挨个完成——如果是，那肯定有同步阻塞混进来了。

5.2 症状：嵌套asyncio.run()，抛RuntimeError

有些同学写了一个异步的主入口，里面又调用了另一个使用asyncio.run()的辅助函数，结果直接报错。原因前面提过，同一个线程的事件循环只能存在一个，asyncio.run()发现当前线程已经有运行中的loop，就不肯再创建新的。

正确的做法是：一个程序只保留一个asyncio.run()入口，内部所有子系统都通过await、create_task或者gather协同工作。如果你在写某个需要被反复调用的工具库，让它提供协程函数就好，由最上层的应用去决定如何调度。

5.3 症状：运行完出现“coroutine was never awaited”警告

这通常是忘记在协程前面写await导致的。忘了await时协程对象永远不会执行，程序虽然不出错，但会留下黄色警告，提醒你某个协程“生而不养”。

我的经验是，只要看到这个警告，就回头检查所有async def的调用点。如果确实不想等待某个协程，说明它应该被asyncio.create_task包装而不是裸调用。这个警告在Python 3.8以后非常显眼，不要忽略它，一般都能顺藤摸瓜找到逻辑漏洞。

5.4 常见问题速查表

结尾

最后分享一点我的个人体会。刚开始学asyncio时，我一度沉迷于把每个函数都改成async，结果代码变复杂了，性能却没提升多少。后来想明白一件事：异步的核心是找到“等待时间占比高”的环节，然后集中精力改造这些环节。爬虫、批量接口调用、文件收集这类场景收益最明显，但普通的小脚本、CPU密集的算法流程，强行改异步只会增加心智负担。

如果你现在手头有一个慢吞吞的同步脚本，建议先别急着全盘重写。挑一个IO最密集的函数，把它改成异步版本，跑一遍对比耗时。当你亲眼看到同样的任务从十几分钟压缩到几十秒，那种满足感会成为你继续深入学习的最大动力。之后再多踩几个坑，多看看Task和超时这些细节，异步编程带来的收益会非常稳定。