说个我自己的经历。前两年帮朋友优化一个数据采集脚本用 Python 写的requests 循环抓几百个页面跑完差不多要二十分钟。我坐在电脑前盯着进度条心里很清楚这二十分钟里 CPU 基本是闲着大部分时间都在等网络返回。也就是从那时候起我开始认真研究 Python 异步编程。如果你也写过类似的同步代码遇到过“明明网络很快程序却慢得像蜗牛”的问题这篇入门教程就是为你准备的。这篇内容不会跟你堆概念而是用最容易理解的类比、最直接的代码示例把 asyncio 的工作原理、核心用法、常见坑位一次讲清楚。看完之后你能独立跑通一个异步爬虫还能明白它为什么快、快在哪里以及哪些场景其实根本不该用异步。无论你是刚入门的小白还是写过一段时间 Python 但一直没搞懂协程的读者都可以放心往下看。1. 为什么要折腾异步同步代码的瓶颈在哪里1.1 从一次真实的爬虫等待说起写爬虫的人几乎都经历过这种场景目标网站并不慢单独请求一个页面也就是零点几秒可一旦写成 for 循环一个接一个抓整体耗时就不是加法而是让人绝望的乘法。下面这段代码是最典型的同步写法import time import requests def fetch_one(url): resp requests.get(url) return len(resp.text) def main(): urls [fhttps://httpbin.org/get?id{i} for i in range(10)] start time.time() total 0 for url in urls: total fetch_one(url) print(f字符总数: {total}) print(f总耗时: {time.time() - start:.2f}s) if __name__ __main__: main()假设每个请求需要 200 毫秒10 个请求串行执行理论耗时就是 2 秒。这 2 秒里程序真正在做的事只有两件发请求、等响应。等待响应的时间占了大头而这段时间内线程完全是空闲的。对于抱着“只要结果能出来就行”心态的人来说这种浪费可以忍但要抓几千个页面、反复调度、频繁重试的场景这种串行等待就是项目里最大的性能杀手。我见过不少新手的第一个多线程爬虫其实也是换汤不换药——线程创建、切换、销毁都要成本如果只是在等待 IO异步通常会比裸开几百个线程更优雅也更可控。1.2 并发、并行、异步这三件事别搞混很多新手把“并发”和“并行”当成一回事结果理解异步时绕了很大的弯。我换个方式说想象一家小咖啡店只有一个店员同时来了五桌客人。并行是指店里雇了五个店员一人服务一桌而并发是只有一个店员但他先记下第一桌的点单转身去收第二桌的杯子再回来给第三桌倒水靠着快速切换把五桌客人都维护住。Python 的 asyncio 就是那个单店员它靠着快速切换任务让单个线程在同一段时间里去照顾多个任务。所谓并行在 CPython 中你绕不开 GIL全局解释器锁真正意义上的 CPU 并行要靠多进程而不是 asyncio。异步擅长解决的是“大量时间花在等待 IO 上”的问题网络请求、文件读写、数据库查询都属于这类。如果一个计算任务纯粹在消耗 CPU异步帮不上太多忙这一点越早想清楚越不容易在选型时走弯路。很多文章把异步写得神乎其神其实剥开来看它就是“把等待的时间挪去干别的活”仅此而已。这一章的目的是帮你在脑子里建立坐标系异步不是银弹它是专门用来收拾“等待浪费”的工具。下面进入原理层看看 asyncio 是怎么把等来的时间偷回来的。2. 事件循环与协程asyncio 的两块基石2.1 协程不是函数是“可暂停的计算”普通函数的执行模型很单纯调用它CPU 一路执行到 return中途不可停。协程不一样它允许你执行到某个位置先暂停把控制权交出去等条件满足时再从暂停点继续。你可以把它想象成一本书读到一半夹了书签回头继续读时不用从头再来而是直接翻开书签那一页。在 Python 里定义一个协程非常简单只要在函数前加 asyncimport asyncio async def hello(): print(协程开始执行) await asyncio.sleep(1) print(协程恢复执行)注意直接调用 hello() 不会执行函数体而是返回一个协程对象。协程对象是可等待的真正驱动它的是事件循环。这里有个非常容易踩的坑新手以为 async def 就是“定义了一个异步函数”调用一下就能用实际上调用后如果没人 awaitPython 只会给你一句 RuntimeWarning提示协程从未被等待。我第一次写异步代码时也犯过这个错误控制台打出一堆警告我当时还以为程序跑坏了。2.2 async/await 到底做了什么把上面代码拆开看。async def 声明了这是一个协程函数协程函数被调用时解释器并不会立刻运行函数体内的代码而是构造出一个协程对象。直到它被扔进事件循环才开始执行。函数体内的 await 是暂停点遇到 await asyncio.sleep(1) 时协程把自己挂起睡眠结束之后事件循环会把它重新唤醒从下一行继续跑。这段机制本质上利用了生成器的“挂起/恢复”能力在 Python 3.4 时代是用 asyncio.coroutine 和 yield from 实现的3.5 之后才提供了 async/await 这套更清晰的语法。理解这一点能帮你读老代码但它对日常写作的意义不大——你只需要记住一条硬规则await 后面跟的一定是可等待对象同时 await 只能在 async 函数里出现。同步函数里写 await 会直接报语法错误。如果你在 REPL 里想试验协程记得用 asyncio.run 或者直接放到一个 async 入口里别想着“我直接调用它试试”。2.3 事件循环像咖啡店店员一样调度一切事件循环是 asyncio 的核心调度器。它维护了一个待办任务队列不断问自己三句话现在有没有可以运行的任务有没有任务在等 IO 且已经等完了有没有任务永久卡死需要超时处理当一个协程 await 一个耗时操作时事件循环不会傻等而是转身去执行队列里的其他协程等那个操作完成事件循环再回来把挂起的协程唤醒。我习惯这样理解事件循环就是一个咖啡店店员。客人点完咖啡店员把订单交给后厨然后去接待下一个客人而不是站在吧台盯着咖啡机。后厨做完咖啡喊一声店员才端起咖啡送给对应的客人。在 asyncio 的世界里“后厨”就是操作系统内核负责通知 IO 事件“客人”就是一个个协程任务。理解了这个模型后面编程时很多直觉都是顺的为什么不能阻塞因为店员一旦被一件事卡住整家店的客人都没人招呼了。3. 手写第一个异步程序从跑通到真正理解3.1 用 asyncio.run 启动你的第一个协程Python 3.7 之后官方推荐用 asyncio.run() 作为入口函数。它会自动创建新的事件循环、运行传入的协程等协程结束后关闭事件循环。写一个最简程序import asyncio async def main(): print(你好异步世界) await asyncio.sleep(1) print(一秒后我才出现) if __name__ __main__: asyncio.run(main())跑一下你会发现程序先打印第一行停顿一秒再打印第二行。这里的 sleep 不是让线程睡觉而是把控制权交还事件循环。虽然这个例子只有一个协程看不出并发的优势但它验证了最关键的一步你已经能让事件循环跑起来了。注意 Python 版本低于 3.7 时 asyncio.run 不存在需要手动用 loop asyncio.get_event_loop() 和 loop.run_until_complete() 来驱动新项目建议直接用 3.8 或更高版本兼容性和便利性都好很多。3.2 可等待对象有三类协程、Task、Futureawait 后面到底能接什么这是初学者问得最多的问题。归纳起来可等待对象有三类协程对象、Task 对象、Future 对象。协程对象直接 await就是按顺序执行Task 是协程被事件循环调度的包装创建后表明“我要并发地跑这个东西”Future 是更底层的概念代表一个尚未完成的结果Task 本质上也是 Future 的子类。可等待对象来源使用场景协程对象async def函数调用后的返回值直接 await 后按序执行Taskasyncio.create_task()包装协程并发调度多个任务Future底层结果容器库开发者直接操作较多实际开发中你打交道最多的就是协程对象和 Task。写代码时如果只是简单 await 一个协程任务仍是串行的只有当多个协程都注册成 Task事件循环才能让它们穿插执行。这就是下一步要验证的事。3.3 用 create_task 让多个协程真正并发看一段有对比效果的代码import asyncio async def say_after(delay, text): await asyncio.sleep(delay) print(text) async def main(): # 串行写法 await say_after(2, 串行你好) await say_after(1, 串行世界) asyncio.run(main())串行版本耗时约 3 秒两个协程一个等完另一个才开始。改成并发版本async def main(): task1 asyncio.create_task(say_after(1, 并发你好)) task2 asyncio.create_task(say_after(2, 并发世界)) await task1 await task2create_task 会把协程立刻注册到事件循环里但不保证立刻执行两个任务同时处于调度队列睡眠结束后谁先到谁先被唤醒总耗时约 2 秒。这里的关键是Task 创建之后你最好记着 await 它否则协程还没跑完事件循环可能就要结束了。另外有一点容易被忽略await task1 和 await task2 的顺序并不代表执行顺序事件循环会按照谁先就绪谁先执行的方式来调度所以打印结果的先后可能和你写的顺序不一致。3.4 一次 gather 并发请求的完整示例create_task 适合手动管理任务但每次都手写一堆 task 再逐个 await代码会很啰嗦。asyncio.gather 就是为批量并发而生的工具。它接收多个可等待对象并发执行返回一个列表顺序和传入顺序一致import asyncio async def fetch(id): await asyncio.sleep(0.5) return fid-{id} done async def main(): results await asyncio.gather( fetch(1), fetch(2), fetch(3), fetch(4) ) print(results) asyncio.run(main())这段代码的总耗时会接近 0.5 秒而不是 2 秒四个协程并行睡眠。gather 的返回值按输入顺序排列不是按完成时间排列写回调或统计结果时不容易乱。需要注意的是gather 默认是“一荣俱荣一损俱损”任何一个任务抛异常其他任务不会被取消但 gather 会把这个异常抛出来。所以你需要在外面包 try/except或者给每个任务单独做异常处理。批量任务里让一个异常毁掉全部结果是我在项目里遇到过的最无语的情况。4. 踩过的五个坑那些文档里不会写的细节4.1 在协程里误用阻塞函数异步瞬间失效这是异步入门最常见的翻车现场。很多人学会了 asyncio写出来的代码也有 async、有 await但看一眼耗时发现和同步没什么区别。最大的嫌疑就是在协程里调用了阻塞函数。time.sleep(2)是最典型的一个。在协程里写await asyncio.sleep(2)事件循环会挂起当前协程转而去执行其他任务但如果写time.sleep(2)整个线程会直接睡 2 秒期间事件循环什么都干不了所有并发任务全部卡住。换成 requests.get、普通的文件 open、数据库的同步驱动效果也一样。它们都是阻塞调用会霸占线程。解决思路有两条。一是换成异步原生库网络请求用 aiohttp 或 httpx 的 async 模式数据库用 asyncpg、aiomysql、motor 这类库文件操作小心对待大文件读写。二是躲不掉阻塞库时把阻塞调用丢进线程池await loop.run_in_executor(None, requests.get, url)让它在另一个线程里执行把执行权及时交还给事件循环。4.2 Task 不保存引用会被垃圾回收你可能写过这样的代码在协程里创建一个 Task既不赋值给变量也不 await然后程序跑完了那个任务里的 print 却一直没有出现。原因是 Task 一旦失去外部引用Python 的垃圾回收会把它当作“不再需要”的对象回收掉事件循环里那个任务还没执行完就被清掉了。正确做法是至少要保存引用task asyncio.create_task(some_coro()) await task如果你希望它在后台静默运行也要用一个列表把 task 存住或者用asyncio.shield()来防止取消。别小看这个细节调试时任务无声无息地消失比报错更难查。我自己就曾经因为这个问题排查了一个多小时最后发现是创建 Task 之后没保存引用任务早就被回收了。4.3 time.sleep 是并发代码里最隐蔽的杀手这个坑值得单独拉出来说。有些场景你需要限速比如爬虫里故意在两次请求之间加延时避免给服务器太大压力。新手常常按同步时代的习惯写time.sleep(1)结果一测量原本设计成并发的代码又变回串行。这里要分清time.sleep 是线程级睡眠整个线程停住asyncio.sleep 是协程级睡眠只挂起当前任务。在异步函数里限速永远优先用await asyncio.sleep()。同样的道理也适用于等待某个条件成立、轮询某个状态凡是需要“等一会儿”的地方都要思考我用的是不是线程级阻塞而不是协程级挂起。我见过有人写轮询逻辑在 async 函数里用了 while time.sleep结果整个服务所有请求都排着队睡觉线上事故现场非常难看。4.4 线程池返回的 Future 必须 await有人一看 run_in_executor 能绕开阻塞问题就到处用它但忘了 await。run_in_executor 返回的是一个 Future 对象它需要被 await 才能真正拿到执行结果。如果你在 async 函数里写下loop.run_in_executor(None, fetch_data, url)然后不 await这行代码只是“提交了任务”并没有执行完返回值也是一个 Future 对象而不是数据。这种代码跑起来经常表现为结果打印不出来程序很快就结束了偶尔还会弹出“Future exception was never retrieved”的警告。处理方式很简单要么result await loop.run_in_executor(...)要么把它包进 create_task 里统一调度。关键是意识到 async 世界里到处都是可等待对象不 await 就相当于下了单但不取货货仓里的东西永远不会自己跑到你手里。4.5 事件循环关闭后还在跑的“幽灵任务”程序主入口写完asyncio.run(main()) 顺利返回但控制台冒出一句 “Task was destroyed but it is pending!”。这说明在事件循环关闭时至少有一个任务还没跑完。最常见的情形是 main 里只 await 了一部分任务另一部分任务因为还在睡眠、还在等 IO被循环关闭时强行取消留下幽灵任务。解决思路是让退出路径足够干净main 里面显式地收集所有任务用 asyncio.wait 或 gather 等它们结束如果需要超时保护用asyncio.wait(..., timeout3)超时后再去 cancel 剩余任务。另外一个好习惯是设置一个统一的取消逻辑在协程的 finally 里做清理避免资源泄漏。想在异步代码里做出“优雅停机”这个坑是绕不过去的。5. 用异步改写真实场景并发 HTTP 请求的完整实战5.1 环境准备为什么选 aiohttp 而不是 requests讲完原理和坑我们上一个真实项目并发请求一组 HTTP 接口并统计返回字符数。这个项目能串起前面所有的知识点。先安装依赖。requests 虽然顺手但它自带的是阻塞式网络 IO在 asyncio 里用不仅享受不到并发还会拖垮事件循环。所以这里选 aiohttp。安装命令pip install aiohttp如果网络较慢可以换用国内镜像源比如pip install -i https://pypi.tuna.tsinghua.edu.cn/simple aiohttp我用的是 Python 3.10 环境aiohttp 装好后直接能用。如果你用的是 conda 环境注意别把 pip 和 conda 的包混在一起装容易出一些莫名其妙的版本冲突。5.2 同步版本先看看它慢在哪为了对比先写一个同步版本。它做的事情非常简单生成 20 个 URL用 requests 逐个请求统计每个响应的 text 长度最后输出总耗时。import time import requests urls [fhttps://httpbin.org/get?id{i} for i in range(20)] def fetch(url): resp requests.get(url) return len(resp.text) start time.time() total sum(fetch(url) for url in urls) print(f同步版本 字符总数: {total}) print(f同步版本 总耗时: {time.time() - start:.2f}s)实测在我本地网络环境下20 个请求大约 4 秒。单独看一个请求也就 200 毫秒20 个串行下来时间几乎就是 200 毫秒乘以 20。每一个请求期间程序都在干等网络响应。如果你把这段代码放在网络波动比较大的环境里耗时会更难看因为任何一个请求变慢后面所有请求都得跟着排队等。5.3 异步版本带限流的完整代码异步版本我加入了一个 Semaphore信号量把最大并发数限制在 10。这样既能显著提速又不会一口气开 20 个连接给服务器压力。信号量的用法我在下一章详细解释这里先看完整代码import asyncio import aiohttp import time async def fetch(session, url, sem): async with sem: try: async with session.get(url, timeout10) as resp: body await resp.text() return len(body) except Exception as exc: print(f请求失败: {url}, {exc}) return 0 async def run(): sem asyncio.Semaphore(10) urls [fhttps://httpbin.org/get?id{i} for i in range(20)] async with aiohttp.ClientSession() as session: tasks [asyncio.create_task(fetch(session, url, sem)) for url in urls] results await asyncio.gather(*tasks) print(f异步版本 字符总数: {sum(results)}) start time.time() asyncio.run(run()) print(f异步版本 总耗时: {time.time() - start:.2f}s)有几个细节值得多说一句。async with aiohttp.ClientSession()是异步上下文管理器它负责连接池的创建和关闭不能换成普通 with。session.get(url, timeout10)里我用的是 aiohttp 自己的 timeout 参数它也是异步语义超时后会取消当前请求而不是阻塞线程。每个任务内部都做了 try/except防止某一个 URL 出错导致整个 gather 崩掉。5.4 实测对比与结果分析在我本机的实测结果同步版本耗时约 4 秒异步版本耗时约 0.6 秒。快不是魔法原理就是那 20 个请求不再排队逐个等而是最多 10 个同时在途等待时间被其他请求填充了。如果网络延迟更高差距会更大如果延迟极低、带宽极高异步优势反而没那么明显。所以评估异步是否值得用时先想想你的场景是不是 IO 密集。这个数字也不建议直接照搬到你自己的环境里因为 httpbin 这类公共服务在不同时间、不同网络下的响应速度差别很大。重点是看趋势IO 密集型任务异步能带来数量级的吞吐提升。如果你在局域网里测自己的服务效果通常更稳定。做性能对比时最好多跑几轮取平均值不要拿一次结果就下结论。6. 更进一步异步编程的正确姿势与能力边界6.1 什么场景适合异步什么场景不适合学到这你应该能判断了。异步最擅长的场景有爬虫批量抓取页面、Web 服务同时处理大量请求、消息队列消费者、API 网关转发、数据库并发查询。这些场景的共同点是大量时间消耗在等待 IO 上等待越久异步收益越明显。反过来异步不适合做纯 CPU 密集型的计算比如大量数值运算、图像处理、加密解密的循环。这类任务线程切换并不会让 CPU 变快单线程里的异步调度甚至会带来额外开销。CPU 密集场景应该用 multiprocessing或者借助 asyncio 的 run_in_executor 把计算丢给进程池。还有一类不适合的是“本来就是简单串行逻辑”的小脚本硬套 async 只会增加阅读成本收益接近于零。6.2 异步不是银弹它不等于跑满 CPU我见过不少同学学完 asyncio 后试图用它解决所有性能问题结果发现 CPU 占用率还是不高程序依然慢。原因很简单asyncio 是单线程调度器它解决的是“等待”的问题不是“计算”的问题。就算你在事件循环里塞了 1000 个并发任务只要任务内容是纯计算CPU 还是得一个个来GIL 也锁着多线程都难并行何况单线程异步。所以正确的心态是把异步当成“时间裁剪器”专门用来回收那些边等边发愁的时间。它和 multiprocessing 是一种互补关系IO 密集用异步CPU 密集用多进程两段逻辑交织时再考虑组合起来用。我见过最离谱的案例是有人用 asyncio 开了 10 万个任务去算斐波那契数列结果比直接同步写还慢就是因为每个任务的调度开销和对象创建开销加到一起反而成了负担。6.3 用 Semaphore 控制并发度防止把远程服务打挂上一章的代码里用了 asyncio.Semaphore。信号量就是并发流量闸门初始化时设定一个最大数量任务在进入关键代码段之前需要先“获取一个许可”用完再“释放”。在爬虫和接口调用的场景里这不是可选项而是必需品无限制地并发可能把对方的服务器打挂也可能让自己的程序因为超时而集体失败。写法很简单sem asyncio.Semaphore(10) async with sem: await session.get(url)把获取信号量的操作放在请求外层能保证同一时刻最多只有 10 个请求在途。调整参数就是调整对目标服务器的压力。我自己的习惯是对小网站并发控制在 5 左右对稳定的 API 服务可以放到 20 到 50具体多少最好先用小批量测试观察错误率和响应时间再来定。一个没人告诉你的细节是Semaphore 本身不是线程安全的但它是协程安全的所以只能在 async 函数里配合 await 使用别拿它去保护多线程代码。6.4 一点维护心得让异步代码好读、好调最后聊点工程上的经验。异步代码最大的问题不是写不出来而是写出来之后难以维护。我在实际项目里一般会遵守三条软规则。第一明确分层把 IO 操作集中在少数几个 async 函数里业务逻辑层尽量少写 await免得整个调用链全是 async。第二任务要有名字和统一出口用 create_task 时给 Task 起个可读的名字比如 task.set_name(fetch-order-123)调试异常时你能从堆栈里一眼看出是哪个任务挂了。第三异常处理要在协程内部完成不要指望“外面包一层 try 就能兜住所有异步异常”因为很多异常是在事件循环的调度间隙被抛出的。调试时有个小技巧设置环境变量 PYTHONASYNCIODEBUG1或者直接 asyncio.run(main(), debugTrue)。事件循环会在任务创建、阻塞调用、未回收 Future 等位置给出详细警告很多隐形坑都能靠它暴露出来。真到了要排查某个协程卡住的时候再配合日志里打印当前所有 Task 的状态基本就能定位问题。这些经验不一定适合所有项目但对我而言它们让异步代码从“跑得动”变成了“看得懂、改得动”。异步编程的乐趣也就在这你终于能把那些碎成片的时间一点一点捡回来。