企业官网建设流程全解析

1. 为什么要拥抱async：从同步爬虫的痛点说起

如果你写过几天爬虫，大概率遇到过这种场景：一个列表页几十个详情链接，用requests挨个串行去请求，每个请求要等网络响应回来才继续下一个，跑起来像乌龟散步。等到想提速，第一反应是开线程池，ThreadPoolExecutor(max_workers=10)，确实快了不少，但线程一多，内存蹭蹭涨，遇到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 = [f"http://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 而非 requests

requests是同步库，虽然好用，但没法在协程里直接用，硬要用就得丢给线程池，那异步的优势就丢了。异步 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, timeout=10) 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, timeout=10) as resp: return await resp.text() except Exception as e: return None

信号量的位置很讲究。如果你把整个协程文件里所有请求都压到同一个信号量下，那是全局限速；如果你按域名分别设置信号量，可以做到不同站点不同并发限额，这就更精细了。我现在一般会在爬虫类里维护一个dict[domain, Semaphore]，按域名动态分配并发额度。

3.4 别忘了优雅退出与超时

真实爬虫跑起来，网络抖一下，某个请求卡住不返回是常有的事。timeout参数如果不设置，你的爬虫可能无限期挂在一个请求上。aiohttp 支持ClientTimeout(total=10)，也可以单独指定连接超时和读取超时。

更复杂一点的场景，比如要爬的数据量大，脚本要跑十几分钟，我需要随时能 Ctrl+C 停掉而不产生僵尸任务。这里推荐用asyncio.wait_for()包裹协程，给每个任务一个硬性超时边界，超时就取消。

try: html = await asyncio.wait_for(fetch_with_limit(session, url), timeout=15) except asyncio.TimeoutError: print(f"请求超时: {url}")

4. 完整实现：从零写一个并发爬虫

原理说得再多，不如直接上一份能跑的代码。我带你把上一节的架构落地，做一个小而全的示例：爬取一个模拟站点的一批详情页，解析标题，写入 JSON 文件。你把这个骨架记熟了，换数据源、加解析逻辑都是顺手的事。

4.1 任务拆分与代码骨架

我的习惯是先定义数据结构。一个待抓取任务长这样：

class FetchTask: def __init__(self, url, retry_count=0): 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, concurrency=5, output="result.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, timeout=10) as resp: if resp.status != 200: raise aiohttp.ClientError(f"bad 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", encoding="utf-8") as f: json.dump(self.results, f, ensure_ascii=False, indent=2) if __name__ == "__main__": urls = [f"https://example.com/page/{i}" for i in range(30)] crawler = AsyncCrawler(urls, concurrency=10) 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_seconds=1): 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_retries=3): for attempt in range(max_retries): try: async with session.get(url, timeout=10) as resp: if 500 <= resp.status < 600: raise aiohttp.ServerError(f"server 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(f"fetch failed for {url}") from e

raise ... from e保留了原始异常链，正好配合日志系统里exc_info=True参数，能打出完整的异常链条，定位问题会快很多。

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 生产消费模型。但所有的进阶玩法，都建立在今天这些基础机制之上。希望这篇文章能让你少踩几个坑，早点写出稳定、高速、可控的异步爬虫。