Python asyncio异步编程实战：从事件循环到并发爬虫性能优化

开头部分

异步编程这两年几乎成了 Python 开发者的必修课，尤其是当你写爬虫、写接口、做数据处理发现程序老是卡在等待 IO 上时，asyncio 就是那把能让你从“一个一个等”变成“同时等一堆”的钥匙。我最早接触 asyncio 是因为一个爬虫脚本要抓几万个页面，用同步 requests 跑要俩小时，后来换成 aiohttp + asyncio 之后直接压缩到十几分钟，那次优化给我留下极深的印象——从此只要碰上 IO 密集型任务，我第一个想到的就是协程。这篇内容我会从 asyncio 最核心的事件循环、协程、任务、await 语法开始讲，配合可以直接跑的代码示例，带你从零搭起一个能用的异步程序。

这篇内容适合谁？已经会用 Python 写点简单脚本、但碰到 asyncio 代码就发懵的入门者；也适合写过同步爬虫或脚本、想搞清楚 async/await 到底是怎么工作的同学。我会尽量不堆术语，遇到绕不开的概念就用生活中的例子拆开讲，保证你看完能自己动手写出第一个异步爬虫或异步请求程序。

主体部分

1. 异步编程到底解决了什么问题

1.1 同步代码的痛点在哪里

先看一个最常见的同步请求例子：你用 requests 库抓 10 个网页，程序会挨个发请求、挨个等响应。每个请求如果耗时 0.5 秒，10 个请求就是 5 秒。这 5 秒里 CPU 大部分时间都在“干等”——数据还没回来，代码只能卡在response = requests.get(url)这一行，什么都做不了。这就是同步阻塞模型的尴尬：CPU 的运算能力被白白浪费在等待上。

如果你抓的页面只有 10 个，这点等待还能忍。可一旦页面数量上到几百、几千，同步循环的耗时就是成倍增长，而且这几乎是线性的。很多新手一开始没有意识到这一点，脚本写到一半才发现速度慢得离谱，然后开始查线程池、查并发，最后绕了一大圈才找到 asyncio。我的建议是：如果任务以网络请求、文件读写、数据库操作为主，也就是所谓的 IO 密集型任务，异步是最值得先试的方案，代码写起来比多线程直观，资源占用也更小。

1.2 异步模型和同步模型的关键区别

同步模型里，函数调用是“一条道走到黑”的：调用某个函数，就必须等它返回才能继续往下走。异步模型则不一样，它允许程序在遇到耗时操作时先“挂起”，把控制权交还给事件循环，然后去执行其他任务；等之前的耗时操作完成了，事件循环再回来接着往下跑。这种机制底层依赖的就是协程，Python 里面用async def定义的函数就是一个协程函数，调用它不会立即执行，而是返回一个协程对象。

很多人会问：这不就是多线程吗？区别其实挺大。多线程是靠操作系统调度线程来切换执行，线程多了会有上下文切换开销，还要考虑共享资源加锁的问题。协程是在单线程内靠事件循环手动切换任务，切换成本极低，而且因为同一时刻只有一个任务在跑代码，普通变量不用加锁也不会有竞争问题（特殊情况除外）。这句话我再说白一点：多线程像是多条流水线同时开工，协程则是一条流水线里，一个人同时管着好几道工序，哪道工序要等待机器，他就先去干别的。

1.3 为什么 IO 密集型任务最适合 asyncio

要理解这个，得先明白 IO 密集型和 CPU 密集型的概念。IO 密集型任务的特点是大量时间花在“等”，比如等网络响应、等磁盘读写、等数据库返回；CPU 密集型任务则是一刻不停地在计算，比如循环计算、复杂算法。asyncio 对 IO 密集型任务特别友好，因为协程让等待时间重叠了——当一个请求在等网络，另一个请求已经开始发了。几个请求的等待时间被叠加在一起，总耗时就约等于最慢的那个请求，而不是所有请求耗时之和。

但如果你的任务是纯计算型的，比如算斐波那契数列算到第 40 项，那用 asyncio 是没用的，因为整个处理过程 CPU 一直忙，根本没有“等待间隙”可以利用。这种场景用多进程或者把计算扔给专用库是更合适的方向。刚开始学 asyncio 的人最容易犯的一个错，就是在 CPU 密集任务上强行上异步，最后发现速度反而更慢，然后得出“asyncio 很垃圾”的错误结论。我想替 asyncio 说句话：不是它不行，是没用在它擅长的赛道上。

2. 环境准备与第一个异步程序

2.1 环境准备建议

从 Python 3.7 开始，asyncio 的 API 已经非常稳定，目前最新版本更是顺手。我建议你用 Python 3.10 或更高版本来学，因为 3.10 之后 asyncio 的报错信息更友好，asyncio.run()这种高级入口用起来也最省心。

先确认一下自己电脑上的环境：打开终端，运行python --version看看版本号。如果版本太低，或者压根没装 Python，那就先去官网下载一个最新版。装完 Python 之后，建议顺手装一个虚拟环境，特别是你手头有多个项目、依赖容易冲突的情况下。虚拟环境建起来很轻松：

python -m venv myenv # Windows 下激活 myenv\Scripts\activate # macOS / Linux 下激活 source myenv/bin/activate

激活虚拟环境后，你在里面装什么库都不会污染全局环境，这一点我强烈建议从入门就开始养成习惯。后续示例中如果用到第三方库，比如aiohttp，就用pip install aiohttp直接装进当前环境。

2.2 第一段可跑的 asyncio 代码

光看理论没用，直接上代码。先写一个最简单的异步函数，感受一下async和await的基本用法：

import asyncio async def say_hello(): print("hello") await asyncio.sleep(1) print("world") asyncio.run(say_hello())

这段代码的逻辑就是：先打印 hello，然后遇到await asyncio.sleep(1)挂起 1 秒，1 秒后继续打印 world。注意，你不能直接调用say_hello()，因为它返回的是一个协程对象，必须通过asyncio.run()或事件循环来运行。

我又写了一个带并发效果的版本，你可以对比一下两段代码执行时间的不同：

import asyncio import time async def task(name, delay): print(f"{name} 开始") await asyncio.sleep(delay) print(f"{name} 结束") async def main(): start = time.time() await asyncio.gather( task("任务A", 2), task("任务B", 1), task("任务C", 3), ) print(f"总耗时: {time.time() - start:.2f} 秒") asyncio.run(main())

asyncio.gather()的作用是把多个协程打包并发执行。输出结果会显示 A 开始、B 开始、C 开始几乎同时发生，结束时则是 B 先结束、A 其次、C 最后，总耗时大约 3 秒而不是 6 秒。这个例子虽然简单，但它就是异步并发的核心体验：等待时间重叠了。

2.3 事件循环是怎么“调度”这些任务的

理解 asyncio，绕不开“事件循环”这个概念。我常用一个餐厅例子来解释它：事件循环就像一个餐厅服务员，他手里有一堆订单（任务）。他接到一个订单后，就把菜下到后厨（发起耗时操作），然后不等这道菜做完，立刻去接其他订单、服务其他客人。后厨的菜做好了，服务员再回来把菜端给对应客人。

在这个例子里，后厨就是操作系统底层的 IO 事件通知机制。协程执行到await时，相当于服务员把工作挂起，操作系统的 IO 完成之后会通知事件循环“那道菜好了”，事件循环再把对应的协程重新拉回运行状态。整个过程没有多线程抢占，全都是在一个线程里排队轮转，所以切换开销极小。

如果你想看看事件循环此刻到底在跑哪些任务，可以用asyncio.all_tasks()获取当前所有任务列表，这在调试并发问题时很有用：

async def main(): task1 = asyncio.create_task(task("任务1", 2)) task2 = asyncio.create_task(task("任务2", 1)) print("当前任务:", asyncio.all_tasks()) await task1 await task2

从输出中你能直观地看到每个任务对象的 id 和状态，这比瞎猜代码跑到哪一步要靠谱得多。

3. 核心 API 与实操细节

3.1 创建任务：create_task vs gather vs ensure_future

实际开发中，很少直接await一个协程对象，因为那样会让程序在等待时没有真正并行执行其他任务。更常见的写法是先把协程包装成任务（Task），再让事件循环调度它们。三种常见方式如下：

• asyncio.create_task(coro)：把协程包装成 Task，立刻交给事件循环调度。注意它只能在事件循环运行时调用，比如在async函数内部。
• asyncio.ensure_future(coro_or_future)：功能类似 create_task，但兼容性更好，可以接受 Future 对象。在较老代码里比较常见，现在官方推荐优先用 create_task。
• asyncio.gather(*coros, return_exceptions=False)：批量并发执行多个协程，并返回所有结果的有序列表。return_exceptions=True时，即使某个任务异常，也不会影响其他任务。

我自己的习惯是：如果用固定的一组协程，优先asyncio.gather，因为它能直接收集结果、写法最简洁。如果是动态生成、数量不定的任务，比如从队列里不断取 URL 并发请求，那就用create_task配合一个列表来管理。来看一个 gather 收集结果的例子：

import asyncio async def fetch_data(id): await asyncio.sleep(1) return f"数据 {id}" async def main(): results = await asyncio.gather( fetch_data(1), fetch_data(2), fetch_data(3), ) print(results) asyncio.run(main())

输出是['数据 1', '数据 2', '数据 3']，顺序和传入顺序保持一致。这个特性在需要保持结果顺序的场景下非常好用。

3.2 超时控制与任务取消

异步程序里最怕的不是“慢”，而是“永久卡住”。网络请求、外部接口随时可能抽风，如果没有超时控制，协程可能会挂在那里不动，整个程序跟着倒霉。asyncio 提供了两个常用工具：asyncio.wait_for()和asyncio.shield()。

wait_for给协程设定一个最长等待时间，超时就直接取消它：

import asyncio async def slow_task(): await asyncio.sleep(10) return "终于完成了" async def main(): try: result = await asyncio.wait_for(slow_task(), timeout=3) print(result) except asyncio.TimeoutError: print("任务超时被取消了") asyncio.run(main())

这段代码里，协程要睡 10 秒，但只给它 3 秒时间，超时后直接抛asyncio.TimeoutError。注意，底层行为不只是抛异常，它还会cancel()掉这个任务，所以如果你对取消逻辑有特殊要求，需要在任务内部捕获asyncio.CancelledError。

shield的含义则是“保护”——它创建一个与内部协程关联的外层包装，当外层任务被取消时，内部协程不会跟着被取消。这在一些需要“即使超时了也不能中断底层任务”的场景很有用，比如你要保存必要的数据、发送最后的上报请求等。我举一个典型例子：某个后台任务正在把结果写进数据库，如果因为调用者取消了任务而中断写入，数据可能不完整，这时候就可以用 shield 包住写入操作。

3.3 控制并发数量：Semaphore 信号量

你可能会想：既然 gather 这么爽，那就一次并发 1000 个任务呗？实际运行起来你大概率会碰到两种后果：一种是目标服务器受不了直接拒绝连接，另一种是自己的程序把文件描述符、内存撑爆。处理这种问题的常规方案是加一个并发上限，用asyncio.Semaphore就能很好地控制。

import asyncio sem = asyncio.Semaphore(5) async def fetch_url(url): async with sem: print(f"开始请求 {url}") await asyncio.sleep(1) print(f"完成请求 {url}") async def main(): urls = [f"https://example.com/{i}" for i in range(20)] await asyncio.gather(*[fetch_url(url) for url in urls]) asyncio.run(main())

这段代码设置了同一时刻最多 5 个协程进入请求阶段，剩下的 15 个协程都在门口排队等待。async with sem这种写法尤其直观，进入代码块就占用一个信号量槽位，退出就释放。我建议在写并发爬虫或批量请求脚本时，不要把并发参数写死在代码里，而是做成命令行参数或配置项，方便随时调整。比如在本地测试用 2-3 个并发就够，跑到服务器上再调大到 20-50，这样能避免一次请求量过大把目标服务打挂。

4. 从同步代码到异步的实战改造

4.1 改造前：一个同步请求版爬虫

纸上谈兵没意思，咱们直接把之前说的“同步爬虫变成异步爬虫”完整跑一遍。先看同步版本，这个代码很简单，用requests循环抓 10 个页面：

import requests import time URLS = [ "https://httpbin.org/delay/1" for _ in range(10) ] def fetch(url): resp = requests.get(url, timeout=5) return resp.status_code start = time.time() for url in URLS: status = fetch(url) print(f"状态码: {status}") print(f"同步版本总耗时: {time.time() - start:.2f} 秒")

httpbin.org/delay/1是一个测试接口，它在返回前强制会睡 1 秒，用来模拟真实网络请求的延迟。同步版本跑完大约要 10 秒以上，因为 10 个请求是一个接一个地等。

4.2 改造后：使用 asyncio + aiohttp

异步版本要用aiohttp这个支持异步的 HTTP 库，先安装再替换请求部分：

pip install aiohttp

import asyncio import aiohttp import time URLS = [ "https://httpbin.org/delay/1" for _ in range(10) ] async def fetch(session, url): async with session.get(url, timeout=5) as resp: return resp.status async def main(): async with aiohttp.ClientSession() as session: tasks = [fetch(session, url) for url in URLS] results = await asyncio.gather(*tasks) for status in results: print(f"状态码: {status}") start = time.time() asyncio.run(main()) print(f"异步版本总耗时: {time.time() - start:.2f} 秒")

异步版本跑完大约只需要 1-2 秒。注意一个细节：ClientSession在并发请求前先创建一次，然后把 session 传给每个 fetch 任务，这样能复用连接池，性能更好。不建议每个请求都单独创建一个 session，那样连接管理反而是浪费。

如果要加并发上限，就把上面的sem = asyncio.Semaphore(5)加进来，在 fetch 里面用async with sem:包住请求部分。实际生产中，这种“信号量 + 并发请求 + 超时控制”的组合几乎是无脑标配，掌握了这一套，大部分爬虫和批量请求场景都能应付。

4.3 为什么性能提升这么明显

有人会问：“异步只是把等待时间重叠了，但总请求数没变，底层网速也没变，为什么不慢反快？”关键在于，大多数机器在单个请求等待时根本没有用它最大的网络带宽。同步代码发一个请求后就傻等，带宽空闲；异步代码同时发起几十个请求，把带宽利用起来了，总体请求完成时间自然缩短。当然，如果目标服务器带宽或你的运营商限制了总带宽，并发再高也不会更快，这时瓶颈就从“等待开销”变成了“网络带宽”。理解这一点，就不会在优化时犯方向性错误。

另外注意，异步并发数量不是越大越好。我自己踩过一个坑：用一个 200 并发的脚本去抓一个小网站，结果对方直接把我的 IP 给封了。所以实用的做法是，并发数从小往大调，观察延迟和错误率再决定上限。对陌生网站，我先用 5 试试；确认没问题了再 20、50 一步步往上加。

5. 常见问题与排查技巧实录

5.1 报错：This event loop is already running

这个报错最常见于 Jupyter Notebook 或某些交互式环境。原因是这些环境本身已经启动了一个事件循环在跑，这时候你在里面直接调用asyncio.run()，就会撞车。解决方案要看场景：

• 如果你用的是 Jupyter，直接await一个协程即可，不需要asyncio.run()。
• 如果你是写脚本，先确认代码顶层没有别的事件循环。
• 如果你确实需要在已运行的循环里再跑一个协程，可以用asyncio.create_task()或者asyncio.get_running_loop().create_task()把它调度进去，而不是再调用一次 run。

这个报错对新手很不友好，因为错误信息不够直白。我第一次在 Jupyter 里跑 asyncio 代码时也卡了半天，最后才搞清楚是“循环套循环”的问题。

5.2 为什么用了 asyncio 还是慢，感觉像同步

这种情况十有八九是因为协程里面调用了同步阻塞函数。比如常见错误写法是：

async def fetch_bad(url): resp = requests.get(url) # 同步阻塞，会卡住整个事件循环 return resp.text

关键点在于，requests.get()是同步阻塞的，它在等待网络时不会让出事件循环，所有并发协程都会在这被卡住，最终效果跟同步代码一样。解决办法是换用异步库，比如 aiohttp；如果必须用同步库，那就需要把阻塞调用丢到线程池里运行，比如用await asyncio.to_thread(requests.get, url)。但这个算是不是特别优雅的办法，优先更换库。

我在代码评审时见过太多这种“假异步”代码了，写法上看着有 async/await，实际跑起来毫无并发效果。这里有个快速判断技巧：看代码里 await 后面跟的是不是原生异步操作，如果是asyncio.sleep、aiohttp.get、async with管理的内容，基本没问题；如果是requests.get、time.sleep、普通文件读取，那多半是假异步。

5.3 调试技巧：开启 asyncio 的调试模式

asyncio 自带一个调试模式，专门用来抓两类问题：协程没有及时被 await 导致的“嘟嘟响”警告，以及任务回调执行时间过长。开启方式很简单：

asyncio.run(main(), debug=True)

或者在事件循环上直接设置：

loop = asyncio.new_event_loop() loop.set_debug(True)

开启调试模式后，Python 会打印一些额外信息，比如有哪些协程对象从未被 await，哪些任务被创建后迟迟没有结果。日志里默认不会打印这些细节，调试模式下则一目了然。有一次我在排查一个程序的“CPU 占用特别高”的问题，就是用 debug 模式发现某个协程里有一个死循环式的密集计算，阻塞了事件循环的正常轮转，修完之后整体流畅了很多。

5.4 常见问题速查表

这张表里的前两个问题是新手最常遇到的，强烈建议收藏一下，等实际报错了再回来对照。

6. 进阶方向与经验之谈

6.1 asyncio 在真实项目里的生态位置

只学 asyncio 本身还不够，得看看它在生产环境长什么样子。以 Python 最火的几个方向为例：

• FastAPI 的异步接口直接把 asyncio 带火了，你用async def定义路径操作函数，FastAPI 自动把它放进事件循环里跑。
• 爬虫框架如 Scrapy 支持协程，某些场景下能明显提升抓取效率；aiohttp、httpx 则是日常自己写脚本时最常用的异步客户端。
• 消息队列消费、WebSocket 长连接、定时任务调度这些场景，也大量依赖 asyncio 的事件循环机制。

也就是说，学会 asyncio 不只是多了一个语法框架，而是打开了异步编程的整个抽屉。后面你学 FastAPI、写并发脚本、做实时数据处理，都会不断地碰到这套概念。

6.2 我的几条实践建议

第一，新手不要在项目一开始就追求复杂的并发结构。先用 gather 把简单的并发跑通，遇到问题再逐步加上信号量、超时、取消这些控制手段。一上来就搞一个庞大的任务调度系统，大概率自己先被搞晕。

第二，理解 async/await 的最好方式是多调试、多打印。把事件循环开启时的顺序、每个任务的开始结束时间都打出来，看多了你就知道协程是怎么切换的。这比反复看文档更有效。

第三，同步代码和异步代码不要混着写。当你用 asyncio 时，尽量全部走异步库，比如文件操作用aiofiles、数据库操作用对应的异步驱动。偶尔混一两个同步阻塞调用看似没事，但在高并发量下会成为隐藏的性能瓶颈。

第四，测试异步代码时，优先用pytest-asyncio这类插件。它允许你把测试函数直接标成异步，然后由框架管理事件循环。十年前很多人因为不知道怎么写异步单测而放弃 asyncio，现在这个门槛已经被磨平了。

第五，多读标准库源码。asyncio 的源码本身写得很清晰，尤其是tasks.py里 Task 的状态转移逻辑，读完你会对“任务是怎么被挂起和恢复的”形成深入理解。当然这一条放到入门的后期再拿，前期先能跑通代码最重要。

最后再分享一个我常用的套路：写任何网络类异步脚本，永远在入口函数外面包一层带try/finally的逻辑，确保所有资源正常关闭、所有任务正常结束。比如用try: await gather包住主要逻辑，finally: await session.close()。别小看这个习惯，它能帮你避开很多“脚本跑完但进程还在挂起”的诡异问题。