Python Asyncio实战：事件循环与协程如何实现高并发爬虫

1. 初识异步编程：Asyncio到底解决了什么问题

做Python开发的人，早晚都会碰上Asyncio这个名字。我看了很多教程，上来就甩事件循环、协程、回调函数，结果新手一脸懵，老手也觉得隔靴搔痒。今天我用实际爬取、请求调度的场景，把Asyncio这套东西掰开揉碎讲一遍，顺便把我在项目中踩过的坑、做过的取舍都交代清楚。

先搞清楚最核心的问题：Asyncio到底在解决什么？一句话概括，它让Python程序在等待I/O时不再干瞪眼，而是先去做别的任务，等I/O完成了再回来继续原来的工作。

传统的同步代码，比如用requests逐个请求接口，每个请求发出后CPU只能等网络响应，这个时间里程序完全“空转”。网络延迟通常几十到几百毫秒，看上去不多，但请求上百个就积累成几秒甚至几十秒的浪费。Asyncio的思路是：请求挂起时，事件循环调度其他协程继续执行，把空闲时间填满，达到“单线程内并发处理大量I/O操作”的效果。

我最初接触这个库时也犯过嘀咕：Python不是有threading线程池吗？多线程不也能并发？确实能，但线程有两个痛点：一是Python全局解释器锁让CPU密集任务并行效果打折扣；二是线程切换开销大，线程多了内存和调度成本都涨，维护起来也费劲。Asyncio是单线程内通过协作式调度完成任务切换，没有线程创建销毁的开销，也没有锁竞争问题，代码结构还更清晰。它适合I/O密集型场景，比如网络请求、文件读写、数据库访问。遇到CPU密集计算，还是老老实实用多进程，这是选型的大前提。

所以这篇文章适合谁？已经被同步代码的性能瓶颈卡住的人、想弄清event loop和协程原理但不想要学院派讲法的人、准备用Asyncio写爬虫或接口调度但不知从何下手的人。看完你会理解事件循环怎么转、async和await到底做了什么、怎么用gather和semaphore控制真正的并发，以及调试协程代码时那些让人抓狂的报错到底什么意思。

2. 核心概念先啃透：事件循环、协程、任务和await

2.1 事件循环是“总调度员”

事件循环是Asyncio的核心引擎，它循环执行的工作是：查看有哪些协程任务可以运行、哪些I/O操作已经完成、哪些任务正在等待。整个流程用大白话说就是——有一堆协程排好队，事件循环挨个检查，能往下走的就走一步，走不了的挂起来，等条件满足再唤醒继续走。这个过程不断重复，程序在宏观上看起来就是“很多任务在同时推进”。

Python 3.7之后，asyncio.run()是启动这个循环最标准的方式。它做的事情是创建一个新的事件循环、运行传入的协程直到它结束、随后关闭循环。日常写代码基本不需要手动管理循环，交给asyncio.run()就好。

import asyncio async def say_hello(): print("Hello, Asyncio!") asyncio.run(say_hello())

这里有个很多人容易忽略的点：async def定义的函数并不会直接执行，调用它返回的是一个协程对象。只有这个协程被交给事件循环运行，函数体才会真正开始执行。看代码时把这一点记住，就不会出现“我调用了异步函数，怎么没打印任何东西”的困惑。

2.2 await关键字：挂起与恢复的标志

await是理解Asyncio的钥匙。它的作用是把控制权暂时交还给事件循环，并告诉循环“我在等某个I/O操作完成，你先去跑别的任务吧”。等到被等待的操作完成，事件循环再回来继续执行await后面的代码。

import asyncio async def fetch_data(): print("开始获取数据") await asyncio.sleep(2) print("数据获取完成") async def main(): await fetch_data() asyncio.run(main())

asyncio.sleep(2)模拟一个需要等待2秒的I/O操作。await asyncio.sleep(2)执行时，事件循环不会傻等2秒，而是去执行其他可运行的协程。2秒后唤醒，接着打印后续内容。理解了这一点，就理解了整个异步编程的运行时模型。

需要注意一个原则：只有async def声明的异步函数内部才能使用await。在普通def函数里写await，直接报语法错误。而且await只能等待可等待对象，比如协程、Task对象、Future对象。刚入门时常见错误是await一个普通函数，Python会提示“object can't be used in 'await' expression”，就是这个原因。

2.3 协程对象与Task的区别

协程对象是函数定义后的“待执行清单”，Task是把这个清单交给事件循环后产生的“正在执行的实例”。这个区分在实战中很关键。

import asyncio async def demo_task(): await asyncio.sleep(1) return "done" async def main(): coroutine = demo_task() # 这只是一个协程对象，还没真正执行 task = asyncio.create_task(coroutine) # 创建Task，注册进事件循环 result = await task # 等待Task执行完成，拿到返回值 print(result) asyncio.run(main())

asyncio.create_task()是创建Task的核心函数，它把协程包装成任务并立即调度到事件循环中。一旦创建，即使你还没await它，任务也会开始运行。这意味着创建多个Task后，它们已经在“并行”推进，后面用await逐一等待结果即可。这也是实现真正并发的基础。

2.4 asyncio.gather与asyncio.wait怎么选

实际项目中经常需要同时发起多个异步任务，等待全部完成后一起处理结果。asyncio.gather是最常用的工具，它接收一组协程或Task，把多个任务变成可以一起等待的Future，返回结果列表。

import asyncio async def fetch_one(url): # 模拟请求 await asyncio.sleep(1) return f"结果: {url}" async def main(): urls = [f"https://example.com/{i}" for i in range(5)] results = await asyncio.gather(*(fetch_one(url) for url in urls)) print(results) asyncio.run(main())

这里的星号非常关键。gather接收的是协程对象或Task作为位置参数，而不是列表。如果直接传列表，会报TypeError。所以用生成器结合星号展开是最常见的写法。

asyncio.wait是另一个选择，但它返回的是两个集合：已完成的任务集合和未完成的任务集合，还需要自己从任务对象里取结果。用起来更繁琐。我的习惯是：只需要全部结果时用gather；需要区分超时任务、手动控制完成条件时用wait。大多数场景gather就够了。

2.5 回调函数与绑定

有些老代码里能看到add_done_callback，它给Task挂一个回调函数，任务完成时自动调用。这种写法能做事，但容易让代码逻辑分散，尤其回调里再嵌套回调，阅读和维护都很痛苦，这就是常说的回调地狱。Asyncio的await语法就是用来取代这种写法的，能把原来靠回调串联的逻辑，改写成从上到下顺序阅读的代码。

async def main(): task = asyncio.create_task(demo_task()) # 任务完成后，回调函数会被自动调用 task.add_done_callback(lambda t: print(f"回调执行: {t.result()}")) asyncio.run(main())

我个人的建议是，新代码尽量别用回调，除非需要把异步任务和某个不使用async语法的事件系统对接。能用await顺序处理，可读性远超回调链。

3. 动手改造：从同步脚本到异步脚本的完整流程

3.1 场景设定与同步版代码

为了说明问题，我设置一个具体场景：需要向5个不同接口发起请求，每个接口延迟大约1到2秒，请求返回后打印响应状态码。先看同步写法。

import time import requests def fetch_url(url): print(f"开始请求: {url}") response = requests.get(url) print(f"完成请求: {url}, 状态码: {response.status_code}") return response.status_code def main(): urls = [ "https://httpbin.org/delay/1", "https://httpbin.org/delay/2", "https://httpbin.org/delay/1", "https://httpbin.org/delay/2", "https://httpbin.org/delay/1", ] start = time.perf_counter() for url in urls: fetch_url(url) print(f"总耗时: {time.perf_counter() - start:.2f}秒") main()

总耗时会接近所有延迟之和，大约7秒。程序绝大多数时间都在等网络响应。这就是典型的I/O密集场景，也是Asyncio大显身手的地方。

3.2 异步改造第一步：换库

要改成异步，首先要把requests换成支持异步的HTTP客户端。最常用的是aiohttp。安装命令为pip install aiohttp。为什么不能直接用requests配合Asyncio？因为requests是同步库，它的网络请求会阻塞当前线程，阻塞期间事件循环无法调度其他协程。即便你写在async函数里，一旦调用requests.get，整个程序的并发就瘫痪了。所以异步改造的第一原则：所有I/O操作必须使用异步版本。

pip install aiohttp

aiohttp的基本用法是创建ClientSession对象，用session.get发起请求，然后用async with确保会话正确关闭。注意它的请求方法和返回内容读取也都是awaitable。

3.3 逐步改造：async/await与gather配合

下面展示一个完整的异步版本改造过程，先拆解每一步再合并成整体。

第一步，把fetch_url改成异步函数：

import aiohttp import asyncio async def fetch_url(session, url): print(f"开始请求: {url}") async with session.get(url) as response: print(f"完成请求: {url}, 状态码: {response.status}") return response.status

这里为什么用session作为参数传入而不是在函数内部创建？因为ClientSession内部维护的是连接池，一个程序创建一个全局会话，比每次请求都新建效率高得多，这是异步HTTP客户端的通用优化点。

第二步，创建会话并批量调度。注意asyncio.gather的传参方式：

async def main(): urls = [ "https://httpbin.org/delay/1", "https://httpbin.org/delay/2", "https://httpbin.org/delay/1", "https://httpbin.org/delay/2", "https://httpbin.org/delay/1", ] async with aiohttp.ClientSession() as session: tasks = [fetch_url(session, url) for url in urls] results = await asyncio.gather(*tasks) print(f"全部状态码: {results}") asyncio.run(main())

第三步，运行并观察结果。5个请求几乎同时开始，整体耗时约2秒，也就是延迟最大的那个请求的耗时。对比同步版的7秒，性能提升非常明显。这个对比是理解Asyncio价值的经典案例。

3.4 为什么不能用threading混用asyncio

有些同学会问：我在async代码里用线程池，把同步库包一层不就行了吗？比如asyncio.to_thread把requests.get丢到线程池里执行，确实能缓解阻塞。但这种做法的本质是“用多线程弥补同步库的缺陷”，事件循环本身没有被充分利用，而且线程数量、锁竞争这些线程问题都回来了。能用原生异步库就优先用原生异步，线程池只是没有替代方案时的兜底手段。

3.5 改造时容易忽略的会话关闭问题

aiohttp.ClientSession不复用或忘记关闭，程序运行一段时间后可能出现连接泄漏，表现是“运行越久越慢”或“文件描述符用尽”。处理方式就是用async with把会话生命周期管理起来。我再强调一遍：session尽量复用，不按请求次数创建。

4. 并发控制与异常处理：实战中的关键细节

4.1 用semaphore限制并发数

gather会把所有任务一次性丢进事件循环。如果任务数量少没问题，但列表里有上百个请求时，瞬间建立大量连接，很容易把目标服务“冲垮”，或者触达本地系统的文件描述符上限。这时必须用信号量控制同时执行的协程数量。这是Asyncio项目必备技能。

信号量的原理是维护一个计数器，每次acquire使计数器减1，计数归零时其他acquire会阻塞等待，release后计数器加1，唤醒等待者。用async with语法即可自动完成acquire和release。

import asyncio import aiohttp async def fetch_with_limit(session, url, semaphore): async with semaphore: async with session.get(url) as response: return await response.text() async def main(): urls = [f"https://httpbin.org/delay/{i % 3 + 1}" for i in range(20)] semaphore = asyncio.Semaphore(5) # 最多同时5个请求 async with aiohttp.ClientSession() as session: tasks = [fetch_with_limit(session, url, semaphore) for url in urls] results = await asyncio.gather(*tasks) print(f"获取到{len(results)}个结果") asyncio.run(main())

很多人会误以为asyncio.Semaphore(5)是“每秒最多5个请求”，实际上它的含义是“事件循环中同时最多存在5个正在运行的协程”。剩余的协程在这里排队，上一个完成后下一个才真正开始执行。这类并发限制对接口服务友好，对自身也防止资源耗尽。真实项目中并发数一般设在10到50之间，具体看你请求的目标性能和本机配置。

4.2 超时控制：别让任务无限等待

网络请求可能永远不返回，不设置超时的异步代码会越跑越不对劲，事件循环里越来越多任务挂起，程序迟迟无法结束。给每个请求设置超时是基本素养。

import asyncio import aiohttp async def fetch_with_timeout(session, url): try: timeout = aiohttp.ClientTimeout(total=5) async with session.get(url, timeout=timeout) as response: return await response.text() except asyncio.TimeoutError: print(f"请求超时: {url}") return None

aiohttp.ClientTimeout(total=5)表示整个连接和读取的总时间上限5秒。另一种做法是用asyncio.timeout或asyncio.wait_for包裹一个awaitable：

async def fetch_with_wait_for(session, url): try: async with session.get(url) as response: return await asyncio.wait_for(response.text(), timeout=5) except asyncio.TimeoutError: return None

wait_for在超时发生时，如果被包裹的Future没有完成会尝试取消它，避免协程一直挂在那里。这两种方式各有适用场景：请求级超时配置适合统一设置所有请求；wait_for适合对某个阶段单独限时。

4.3 异常隔离：一个报错不能拖垮所有任务

gather有个容易踩的坑：默认情况下，如果其中任何一个任务抛出未捕获异常，gather会直接抛出异常，可能导致整个程序中断，即便其他任务正常完成了。解决手段是return_exceptions=True。

results = await asyncio.gather(*tasks, return_exceptions=True)

这样任务异常会作为结果返回，而不是向上抛出。比如发生网络错误时，结果是Exception对象。处理结果时用isinstance判断一下类型再决定怎么取值。这个参数我在项目中几乎每次都带上，它让“部分失败”的场景变得可控制。

4.4 结构化的异常处理方式

更稳妥的实践是在每个协程内部捕获异常，返回一个带状态的结果对象。比如用字典统一结构：

async def fetch_with_result(session, url): try: async with session.get(url) as response: return {"url": url, "status": response.status, "data": await response.text()} except Exception as exc: return {"url": url, "status": None, "error": str(exc)}

之后在汇总阶段统一筛选成功和失败的任务，重试失败项。这种模式比大量try/except嵌套更清晰，也方便统计整体成功率。处理数十个并发任务时，代码不会乱。

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

5.1 RuntimeError: no running event loop

初学者最常见的一个报错是“RuntimeError: no running event loop”。原因通常是直接在模块顶层调用了create_task或get_event_loop相关操作，但那时事件循环还没启动。解决方法是把所有异步代码包进async main()，用asyncio.run(main())启动。别在async函数外面使用create_task，这是写异步代码的基本纪律。

5.2 “Event loop is closed”是什么鬼

代码执行完毕后再去访问事件循环相关资源，可能出现“Event loop is closed”。常见场景：全局创建了Task但没有保存引用，程序结束时事件循环已经关闭，任务却被垃圾回收或试图继续执行。解决思路是确保所有Task在程序结束前都已经被await完成，或者在close前手动cancel未完成的任务。asyncio.run会负责在结束时关闭循环，但你代码里泄漏的Task仍是隐患。

5.3 请求结果乱序？注意任务完成顺序不一定按创建顺序

gather返回的结果顺序和传入的任务顺序一致，这一点方便了对号入座。但如果你直接用列表逐个创建Task再依次await，比如：

tasks = [] for url in urls: task = asyncio.create_task(fetch_url(session, url)) tasks.append(task) # 等一下，错误示范来了 for task in tasks: print(await task)

注意create_task之后任务会立即被调度执行，await是等待结果而不是“启动”任务。如果你在循环里先创建所有任务再统一await，这不是问题。真正的坑是在循环里边创建边await，那会退化成串行等待。

# 错误示范：边创建边await会变成串行 for url in urls: task = asyncio.create_task(fetch_url(session, url)) await task # 这会等一个完成才开始下一个

写成这样，每次循环都创建一个任务然后立刻等它完成，异步并发效果就完全丢失了。排查这类问题时，重点检查是否把create_task放在循环外、把await放在循环外。

5.4 每个请求都创建新Session，性能差到离谱

有个现象：程序总耗时比预期长很多，CPU还飙高。原因往往是吧ClientSession的创建写在了单次请求内部，每次请求都新建连接、重建连接池。遇到这种问题，第一件事就是检查Session是不是在循环外创建的。Session是重量级对象，必须复用。

5.5 调试协程代码的实用技巧

asyncio提供了debug模式，通过asyncio.run(main(), debug=True)开启。调试模式下，事件循环会输出未完成任务的告警，也会显示回调函数执行耗时超过规定阈值的信息，对查找任务泄漏非常有用。

另外一个实际调试技巧：在关键协程入口打印时间戳，辅助观察任务实际并发情况。比如：

import asyncio import time async def worker(name): print(f"{time.time():.3f} 任务{name}开始") await asyncio.sleep(2) print(f"{time.time():.3f} 任务{name}结束")

看输出里相邻任务“开始”时间差，就能知道它们是否并发执行。

5.6 快速排查清单

这张表是我排查时最常翻的参照，新手遇到类似问题可以直接对号入座。

6. 从入门到精通的进阶路线建议

掌握了基础和实战之后，继续深入的方向主要有三个。

第一个方向是深入理解底层事件循环。asyncio的异步能力建立在selectors和epoll这些操作系统I/O多路复用机制上。理解了epoll是基于事件通知的回调驱动模型，就能明白为什么网络I/O等待不占用CPU。像asyncio.Protocol和asyncio.Transport这类底层接口，虽然不常用，但读一遍源码能加深对网络编程的认知。

第二个方向是学习asyncio与其他并发模型协作。有些项目用Asyncio处理高并发网络层，但遇到CPU密集计算时会配合concurrent.futures.ProcessPoolExecutor把任务交给子进程。跑满多核的同时保持主流程便携复用。多模型配合才是真实分布式系统的常态。

第三个方向是生态库积累。除了aiohttp，很多数据库驱动都有异步版本。比如面向PostgreSQL的asyncpg，面向Redis的redis.asyncio。就连ORM也有SQLAlchemy的Asyncio扩展。掌握了基础后，把这些异步驱动组合起来，就能写出全套异步的服务端程序，此时你再看早期那些同步代码，会觉得处处都是性能浪费点。

我个人实际使用中最深的体会是：Asyncio的学习曲线不是坡度陡，而是拐弯多。理念一句话能说清，但真正用来写工程代码时，每个细节都有可能变成坑。想把异步用好，重点不是背API，而是建立“事件循环视角”——时刻问自己“当前这行代码运行到await时，控制权交给了谁，其他任务处于什么状态”。多问几次，很多设计上的困惑就自然消解了。

最后再分享一个小技巧：写异步代码时，尽量把可等待对象的结构设计成一目了然。比如统一返回结果字典，统一异常兜底。表面上看是多写了几行逻辑，实际运行中排查问题会痛快得多。异步编程最难的部分从来不是语法，而是工程化的组织能力，这部分功夫只能靠实打实地写代码慢慢攒。