★ FEATURED ARTICLE 1. 为什么要拥抱async从同步爬虫的痛点说起如果你写过几天爬虫大概率遇到过这种场景一个列表页几十个详情链接用requests挨个串行去请求每个请求要等网络响应回来才继续下一个跑起来像乌龟散步。等到想提速第一反应是开线程池ThreadPoolExecutor(max_workers10)确实快了不少但线程一多内存蹭蹭涨遇到requests卡住还没超时的情况整个线程池被僵住的任务占满调度又开始乱套。我早期写爬虫就是这个状态串行太慢多线程费力。后来接触到 asyncio彻底换了思路它不是靠多线程去并行执行而是在单线程内通过事件循环调度协程把等待网络响应的空闲时间用来干别的事。这个概念听起来玄乎实际用起来就一句话遇到 I/O 操作比如发 HTTP 请求就主动让出控制权等数据回来了再回来继续执行。Python 的异步生态这几年已经很成熟了asyncio是标准库自带网络请求有aiohttp、httpx这些三方的异步客户端配合lxml、BeautifulSoup做解析写一个几十路并发的爬虫并不需要动多少代码。这篇文章适合谁看大概两种人一种是把 Python 同步爬虫写得滚瓜烂熟、但还没搞懂 asyncio 的另一种是看过不少 asyncio 教程但拿它做真实项目时总踩坑的。我会从事件循环和协程的运行机制讲起然后带着你从零实现一个带并发控制、超时重试、失败降级的异步爬虫Intel Mac 和 Windows 环境我会一并提醒注意事项。先给你一个准话异步不是银弹它本质是为 I/O 密集型场景优化的解析 HTML 这类 CPU 密集的工作不能丢进协程里死等。理解了边界你才能在对的场景把它用得顺手。2. 事件循环与协程asyncio 的底层运转逻辑我在刚开始学 asyncio 时最困惑的一件事是async def定义出来的函数为什么调用它不会执行函数体而是返回一个协程对象后来把事件循环想明白这个疑问就消失了。2.1 事件循环到底是什么你可以把事件循环想象成一个超级高效的调度员它维护着一个待办队列。协程执行到await时会把控制权交还给调度员调度员一看这个协程在等 I/O就把它挂起接着从队列里拉下一个协程出来跑。等之前的 I/O 完成了调度员收到通知再把这个协程放回队列让它从 await 那行继续往下走。这个机制的关键在于单线程内通过不断的切换让 CPU 和网络通道始终处于忙碌状态。因为 HTTP 请求的大部分时间都浪费在网络延迟上传统同步模型里线程在那干等异步模型里这个等待被调度员回收利用去执行其他协程。Python 里的asyncio.run()是 3.7 之后推荐的入口它会创建新的事件循环执行传入的协程结束后再关闭循环。常规代码长这样import asyncio async def main(): print(hello) await asyncio.sleep(1) print(world) asyncio.run(main())注意asyncio.sleep(1)不是卡住线程 1 秒而是让出控制权事件循环这 1 秒里可以去跑别的协程。这个细节是理解之后所有用法的基石。2.2 协程与 Task 的区别很多人在这里栽跟头async def定义的函数调用后返回的是协程对象它本身不会运行。要让它们真正被调度有两个常见方式一是直接await coro也就是在当前协程里等待它完成二是用asyncio.create_task(coro)把它包装成Task对象Task 会被立即安排到事件循环里但执行是异步的你可以同时创建很多 Task它们会一起被调度。用一个例子就能看清区别import asyncio async def say_after(delay, message): await asyncio.sleep(delay) print(message) async def main(): # 方式一串行等待总耗时 3 秒 await say_after(1, first) await say_after(2, second) # 方式二创建两个任务总耗时 2 秒 task1 asyncio.create_task(say_after(1, task1)) task2 asyncio.create_task(say_after(2, task2)) await task1 await task2 asyncio.run(main())我这里特别强调一下很多人写爬虫时误以为async with session.get(url) as resp里套个await就万事大吉其实你如果只是在main()里挨个 await那依然是串行的。真正的并发一定要靠 create_task 批量创建任务或者用 asyncio.gather/asyncio.wait 去聚合。2.3 gather、wait、TaskGroup 怎么选旧代码里最常看到的是asyncio.gather它会同时调度多个协程返回所有结果。如果其中一个抛异常默认会向上抛出导致其他任务即使完成了也不等结果。asyncio.wait则更底层让你能通过return_when参数控制返回时机比如只要有一个完成就先回来处理。Python 3.11 之后引入了asyncio.TaskGroup这玩意儿比 gather 好用得多。它结构上更清晰异常处理也更安全会在所有任务结束后统一抛错避免任务泄露问题。我现在的项目基本都改用 TaskGroup 了但考虑兼容老代码下面的爬虫实例我会先用 gather 演示再给你一个 TaskGroup 版本让你对比。import asyncio async def fetch_url(url): await asyncio.sleep(1) return url async def main(): urls [fhttp://example.com/{i} for i in range(5)] # gather 版本 results await asyncio.gather(*(fetch_url(u) for u in urls)) print(results) # TaskGroup 版本 async with asyncio.TaskGroup() as tg: tasks [tg.create_task(fetch_url(u)) for u in urls] print([t.result() for t in tasks]) asyncio.run(main())这段代码跑起来你会发现两种写法的总耗时都只有约 1 秒假设每请求 sleep 1 秒因为它们是并发执行的。爬虫的思路从这里就通了把每个 URL 的请求封装成一个协程批量创建任务事件循环帮我们并发调度不再需要手动管线程。3. 设计一个可复用的异步爬虫架构写异步爬虫和写同步爬虫有个很重要的区别同步代码你可以想到哪写到哪异步代码如果一开始不规划好任务模型后面改起来特别痛苦。我第二次写异步爬虫时就吃过这个亏逻辑乱成一团信号量还加错位置导致同一批请求冲出去了几十个被对方站点直接封了 IP。3.1 明确爬虫的四个阶段一个规矩的爬虫无论同步异步都逃不过这四步请求、解析、存储、调度。异步爬虫的区别在于请求阶段是并发密集的解析阶段如果量大要分流存储阶段尽量走异步或者批处理。我习惯用生产者-消费者模型去组织代码。生产者负责从待抓取队列里取出 URL交出协程去请求请求完成后解析器把 HTML 转成结构化数据消费者负责把数据写进文件或数据库。asyncio 里的asyncio.Queue天然支持协程并发读写比用queue.Queue加线程要自然得多。结构大致是这样一个 URL 队列存放待抓取地址若干 worker 协程各自循环从队列取 URL、请求、解析、存储一个主协程负责填充初始 URL 并等待所有 worker 完成3.2 为什么我推荐 aiohttp 而非 requestsrequests是同步库虽然好用但没法在协程里直接用硬要用就得丢给线程池那异步的优势就丢了。异步 HTTP 客户端里我主力用的是aiohttp它的ClientSession底层维护了连接池能复用 TCP 连接对比每次新建连接性能提升明显。httpx也是好选择它的 API 比 aiohttp 更贴近 requests写起来顺手还支持 HTTP/2。不过httpx的异步客户端有个坑它默认使用多线程底层备选机制对某些老版本urllib3可能会有兼容问题如果你只是想稳妥地跑起来aiohttp的生态更成熟文档也更全。import aiohttp import asyncio async def fetch(session, url): async with session.get(url, timeout10) as resp: return await resp.text() async def main(): async with aiohttp.ClientSession() as session: html await fetch(session, https://example.com) print(len(html)) asyncio.run(main())ClientSession记得要复用不要每个请求都创建新 session。session 内部维护的连接池和 cookie 状态是跨请求共享的频繁开关会浪费连接还可能触发反爬。3.3 并发控制信号量告诉你什么叫克制并发爬虫最大的风险不是写不出来而是写出来之后把目标站点打挂或者被 WAF 拦截。asyncio.Semaphore就是干这个的它限制同一时刻最多允许多少个协程进入关键区。import asyncio import aiohttp semaphore asyncio.Semaphore(5) async def fetch_with_limit(session, url): async with semaphore: try: async with session.get(url, timeout10) as resp: return await resp.text() except Exception as e: return None信号量的位置很讲究。如果你把整个协程文件里所有请求都压到同一个信号量下那是全局限速如果你按域名分别设置信号量可以做到不同站点不同并发限额这就更精细了。我现在一般会在爬虫类里维护一个dict[domain, Semaphore]按域名动态分配并发额度。3.4 别忘了优雅退出与超时真实爬虫跑起来网络抖一下某个请求卡住不返回是常有的事。timeout参数如果不设置你的爬虫可能无限期挂在一个请求上。aiohttp 支持ClientTimeout(total10)也可以单独指定连接超时和读取超时。更复杂一点的场景比如要爬的数据量大脚本要跑十几分钟我需要随时能 CtrlC 停掉而不产生僵尸任务。这里推荐用asyncio.wait_for()包裹协程给每个任务一个硬性超时边界超时就取消。try: html await asyncio.wait_for(fetch_with_limit(session, url), timeout15) except asyncio.TimeoutError: print(f请求超时: {url})4. 完整实现从零写一个并发爬虫原理说得再多不如直接上一份能跑的代码。我带你把上一节的架构落地做一个小而全的示例爬取一个模拟站点的一批详情页解析标题写入 JSON 文件。你把这个骨架记熟了换数据源、加解析逻辑都是顺手的事。4.1 任务拆分与代码骨架我的习惯是先定义数据结构。一个待抓取任务长这样class FetchTask: def __init__(self, url, retry_count0): self.url url self.retry_count retry_count然后核心爬虫类大概就是这三块初始化 session 和信号量、定义单个页面抓取流程、定义 worker 主循环。4.2 核心代码逐行拆解import asyncio import aiohttp import json import time from bs4 import BeautifulSoup class AsyncCrawler: def __init__(self, urls, concurrency5, outputresult.json): self.urls urls self.output output self.semaphore asyncio.Semaphore(concurrency) self.results [] self.lock asyncio.Lock() async def fetch(self, session, url): async with self.semaphore: async with session.get(url, timeout10) as resp: if resp.status ! 200: raise aiohttp.ClientError(fbad status {resp.status}: {url}) return await resp.text() async def parse(self, html): soup BeautifulSoup(html, lxml) return { title: soup.title.string.strip() if soup.title else , url_count: len(soup.find_all(a)) } async def process(self, session, url): try: html await self.fetch(session, url) data await self.parse(html) # 用锁保护共享列表写入避免多个协程同时 append 出乱子 async with self.lock: self.results.append({url: url, **data}) except Exception as e: print(f失败: {url}, 错误: {type(e).__name__} {e}) async def worker(self, session): while True: try: url self.urls.pop() except IndexError: break await self.process(session, url) async def run(self): async with aiohttp.ClientSession() as session: workers [asyncio.create_task(self.worker(session)) for _ in range(3)] await asyncio.gather(*workers) with open(self.output, w, encodingutf-8) as f: json.dump(self.results, f, ensure_asciiFalse, indent2) if __name__ __main__: urls [fhttps://example.com/page/{i} for i in range(30)] crawler AsyncCrawler(urls, concurrency10) asyncio.run(crawler.run())几个要注意的细节asyncio.Lock不是线程锁不能跨线程用但在这个单线程事件循环里是安全的。多个协程同时调用results.append没有锁保护的话虽然 GIL 保证单个字节码不会被穿插但复杂对象写入还是建议加锁保险。self.urls.pop()在多 worker 场景下会竞争取 URL虽然这里只是演示但如果你用的是asyncio.Queue要用await queue.get()而不是pop因为Queue.get()本身是协程安全的。asyncio.gather(*workers)如果没有捕获异常任何一个 worker 崩溃会影响到主流程。所以我在process里捕获了所有异常宁可记录失败也不让整个脚本挂掉。4.3 产量和性能实测拿上面的代码在本地网络环境爬 30 个页面并发限制 10总耗时大概在 3 到 5 秒之间。同样的任务用同步 requests 串行轻松 15 秒以上。你可能会问线程池能做到差不多速度吗答案是能但线程池的问题在于线程切换开销、内存占用、以及线程和 asyncio 混用的复杂度。当你后续要加限速、重试、持久化、失败恢复这些功能时同步线程模型的复杂度会成倍增长asyncio 的显式协程控制反而更清晰。4.4 扩展点把存储改造成批处理上面的代码每个页面解析出来就立刻写到共享列表最终统一落盘。真实项目里如果数据量是百万级你不能全塞在列表里等结束再写内存会爆掉。建议改成攒批写入解析结果塞进一个缓冲区每满 100 条就异步写一次数据库或文件。buffer [] async def flush_buffer(buffer, output_writer): if len(buffer) 100: return await output_writer(buffer) buffer.clear()好处很明显写磁盘/数据库是 I/O 操作单独拎出来和抓取并发是合理的但一旦批量化了写操作的频率降下来磁盘 I/O 压力也会小很多。5. 限速、重试与失败恢复生产级爬虫的必经之路把小爬虫放大到每天跑几万条 URL 的生产环境光有并发还不够你需要在稳定性上多花功夫。我在做某数据采集项目时就踩过不少坑这里把最重要的几点给你提炼出来。5.1 限速不是简单的 sleep上一节用 Semaphore 控制并发数量但很多站点是按时间窗口限频的比如每秒最多 5 个请求。光靠并发上限做不到这一点。这时候需要令牌桶思路维护一个下一次允许请求的时间戳每个 worker 在发起请求前都检查当前时间是否达到这个时间戳没到就睡到那个点。class RateLimiter: def __init__(self, rate, per_seconds1): self.min_interval per_seconds / rate self.next_allow asyncio.get_event_loop().time() async def wait(self): now asyncio.get_event_loop().time() if now self.next_allow: await asyncio.sleep(self.next_allow - now) self.next_allow self.min_interval把这个 limiter 塞进fetch()里的信号量之后请求顺序就有节奏了。鉴权站点、电商平台、新闻门户对频率的感知各不相同你要学会根据响应头里的Retry-After、状态码 429 动态调整限速参数。5.2 重试机制的几个坑网络请求失败不可怕可怕的是无脑重试把站点打得更狠。我设计的重试原则是只对幂等请求重试GET 天然安全POST 就要谨慎必须使用指数退避且加一点随机抖动重试次数上限 3 次超过就放弃记录日志async def fetch_with_retry(session, url, max_retries3): for attempt in range(max_retries): try: async with session.get(url, timeout10) as resp: if 500 resp.status 600: raise aiohttp.ServerError(fserver error {resp.status}) return await resp.text() except Exception as e: if attempt max_retries - 1: raise backoff 2 ** attempt random.uniform(0, 1) await asyncio.sleep(backoff)注意random.uniform(0, 1)的抖动非常关键没有抖动的指数退避会导致所有客户端同步重试形成重试集结效应反而把服务端打垮。5.3 失败数据不丢弃断点续爬真实爬虫经常会中途挂掉你总不能每次都从头爬。我在跑长任务时会把已完成 URL 集合定期快照到磁盘启动时先加载快照跳过已经成功的 URL。快照文件用 Python 的pickle或 JSON 都行每处理完一批就更新一次。done_set set() checkpoint_file done_urls.json # 启动时加载 if os.path.exists(checkpoint_file): done_set set(json.load(open(checkpoint_file))) # 每完成一个 URL 就加入集合 done_set.add(url) # 定期保存 if len(done_set) % 50 0: json.dump(list(done_set), open(checkpoint_file, w))配合这个方法即使脚本崩了重启之后能很快接着跑不会造成大量重复请求。5.4 日志是调试法医异步爬虫一旦出了诡异问题print 大法基本没用因为多个协程交错打印根本看不出来先后顺序。用logging模块带线程号、时间戳、协程名输出会清晰很多。我通常在每个请求开始和结束时各打一条日志记录 URL、状态码、耗时排查慢请求和失败请求时就靠它们。6. 我在这条路上踩过的三个大坑有些坑是教程不会写、只有自己跑挂了才明白的。我把印象最深的三个记录下来你遇到时可以少走弯路。6.1 在协程里混用了同步阻塞库这是新手最容易犯的错。比如在协程里直接调用requests.get()它会把事件循环卡死其他协程全停在那里等这个请求返回。你以为自己在写异步实际上把异步变成了更糟糕的串行伪并发。诊断方法很简单如果并发 10 路总耗时是单请求耗时的 10 倍以上基本就是有人在协程里干阻塞 I/O 了。我在项目里用time.sleep()做测试时也踩过同样的问题记住协程里要用await asyncio.sleep()不能用time.sleep()。合规的做法是如果你非要调同步库就把它丢给线程池loop asyncio.get_event_loop() result await loop.run_in_executor(None, requests.get, url)但这只是在没法改库时的妥协最佳方案还是用 aiohttp 这类原生异步库。6.2 异步调试比你想的要难异步代码的堆栈追踪非常混乱。一个异常可能在某个 3 层深的任务里抛出但堆栈却回溯到asyncio.run()的调用处中间过程的变量值早被覆盖了。我的调试策略是把关键异常包装得更细致try: html await fetch(session, url) except aiohttp.ClientError as e: raise RuntimeError(ffetch failed for {url}) from eraise ... from e保留了原始异常链正好配合日志系统里exc_infoTrue参数能打出完整的异常链条定位问题会快很多。6.3 忘掉 asyncio 的全局唯一事件循环这件事不同版本的 Python 对事件循环的管理机制不太一样加上 jupyter notebook 环境、某些测试框架内部的 event loop很容易引发RuntimeError: asyncio.run() cannot be called from a running event loop这种经典报错。解决办法是给一个入口函数做环保检测def launch(main_coro): try: asyncio.get_running_loop() except RuntimeError: asyncio.run(main_coro) else: # 已经运行在事件循环里直接创建任务 return asyncio.create_task(main_coro)这样在脚本模式和 Jupyter 模式里都能正常运行不会因为环境差异炸掉。7. 收尾把异步编成肌肉记忆异步编程不是换一门新语言更像是换一种思考 I/O 的方式。我写了几年异步代码最大的体会是先习惯让出控制权再谈并发速度。你不需要一上来就背一堆 API把await、create_task、Semaphore、Queue这几个核心概念用熟能覆盖 90% 的爬虫场景。最后分享一个我很受用的小技巧写异步代码之前先在纸上画出三条时间线。一条是 CPU 在做什么一条是网络请求在做什么一条是事件循环在调度什么。画着画着你就会发现很多为什么这里要 await、为什么那里要 create_task的问题瞬间有了答案。这个项目的扩展空间还有很多比如把上面的爬虫骨架套进 scrapy 的异步中间件、用aiofiles做全链路异步文件写入、结合asyncpg做异步 PostgreSQL 存储甚至可以在协程里优雅嵌套 RabbitMQ 生产消费模型。但所有的进阶玩法都建立在今天这些基础机制之上。希望这篇文章能让你少踩几个坑早点写出稳定、高速、可控的异步爬虫。
阅读完成 · 觉得有帮助？
