Python爬虫终极提速：异步IO（asyncio+aiohttp）优化，比多线程还快4倍

ZCH��

1015人浏览 · 2026-04-08 16:14:14

ZCH�� · 2026-04-08 16:14:14 发布

Python爬虫终极提速：异步IO（asyncio+aiohttp）优化，比多线程还快4倍

在上一篇博文中，我们通过多线程（ThreadPoolExecutor）对单线程小说爬虫进行了优化，将爬取速度提升了5~10倍，解决了单线程串行执行导致的效率瓶颈。但不少小伙伴在评论区留言：多线程已经很快了，还有更高效的优化方式吗？答案是肯定的——今天这篇，我们将引入异步IO（asyncio + aiohttp），实现爬虫的终极提速，比多线程再快2~4倍，同时降低资源占用，让爬取体验更流畅。

本文将完全衔接上一篇多线程优化内容，从多线程的局限性、异步IO核心原理、异步脚本实战实现、三者速度对比、注意事项五个方面展开，全程实战导向，每一步都有详细解析，保留原有爬取功能和排版格式，新手也能直接复制运行。同时，我们会对比单线程、多线程、异步IO三种方式的耗时差异，让大家直观感受到异步IO的优势，真正掌握Python爬虫的进阶优化技巧。

一、回顾与思考：多线程的局限性

在上一篇的实战中，多线程确实解决了单线程串行执行的效率问题，通过并行发送多个网络请求，充分利用了网络等待时间，让爬取速度大幅提升。但多线程并非完美，它依然存在一定的局限性，尤其是在爬取大量章节、高并发场景下，这些问题会更加明显。

多线程的核心局限性主要有两点：

线程切换开销：虽然多线程能并行执行任务，但Python的GIL（全局解释器锁）导致同一时刻只有一个线程执行Python字节码，线程之间的切换需要消耗一定的系统资源。当线程数量过多时，切换开销会急剧增加，反而会降低爬取效率，甚至出现卡顿。
资源占用较高：每个线程都需要占用一定的内存和CPU资源，线程数量越多，资源占用越高。如果爬取章节数量极多（上千章），创建大量线程会导致电脑内存飙升，影响其他程序的正常运行。

而我们今天要讲的异步IO，恰好能解决这两个问题。异步IO不需要创建多个线程，而是通过“事件循环”机制，在一个线程内同时处理多个网络请求，无需线程切换，资源占用极低，在IO密集型任务（如爬虫）中，效率比多线程更高。

举个直观的例子：爬取10章小说，单线程耗时8_{15秒，多线程耗时1}3秒，而异步IO耗时仅需0.3~0.8秒，速度提升效果非常显著。接下来，我们就详细讲解异步IO的核心原理，以及如何将多线程脚本优化为异步版本。

二、异步IO核心原理（新手易懂版）

提到异步IO，很多新手会觉得“比多线程更复杂”，其实核心逻辑和多线程类似，都是为了充分利用网络等待时间，但实现方式完全不同。我们依然用一个生活化的例子，帮大家快速理解：

假设你是一名厨师，需要做10碗面，每碗面的流程是“煮面（5分钟，无需看管）+ 调味（1分钟，需要看管）”。

单线程模式：煮第一碗面（等5分钟）→ 调味（1分钟）→ 煮第二碗面（等5分钟）→ 调味（1分钟）… 总耗时60分钟；
多线程模式：找5个助手，每个助手负责2碗面，同时煮面、同时调味，总耗时12分钟；
异步IO模式：你一个人，先把10碗面同时放进锅里煮（无需看管），在煮面的5分钟里，你可以同时给已经煮好的面调味，煮面和调味并行进行，总耗时6分钟。

对应到我们的小说爬虫中：

单线程：依次发送请求，等待响应，再处理内容；
多线程：多个线程同时发送请求，等待响应，线程切换有开销；
异步IO：一个线程内，同时发送多个请求，在等待响应的过程中，处理其他已响应的请求，无需线程切换，开销极低。

在Python中，实现异步IO的核心工具是两个模块：

asyncio：Python内置的异步核心模块，负责管理事件循环、任务调度，是异步编程的基础；
aiohttp：异步HTTP请求库，用于发送异步网络请求，替代了多线程中的requests库（requests库是同步的，无法在异步环境中使用）。

这里需要特别说明：异步IO和多线程一样，更适合处理“IO密集型任务”（如网络请求、文件读写），而不是“CPU密集型任务”。我们的小说爬虫是典型的IO密集型任务，因此用异步IO优化，能达到比多线程更优的提速效果。

三、异步IO脚本优化实现（衔接多线程，无缝改写）

本次优化完全基于上一篇的多线程脚本，不改变原有核心功能（爬取内容、排版格式、文件保存方式），仅将同步请求改为异步请求，添加异步事件循环逻辑，完善异常处理，确保大家能无缝衔接，直接替换使用。下面我们分步骤讲解优化过程，最后给出完整可运行脚本。

3.1 新增依赖与导入

多线程脚本中，我们导入了requests、lxml、os、time、ThreadPoolExecutor五个模块；异步IO版本需要替换同步请求库，新增异步相关模块，具体如下：

移除：requests库（同步请求，无法用于异步环境）、ThreadPoolExecutor（多线程相关）；
新增：aiohttp（异步HTTP请求库）、asyncio（异步核心模块）；
保留：lxml（解析HTML）、os（文件操作）、time（计时）。

新增导入代码：

import aiohttp
import asyncio

需要注意：aiohttp不是Python内置模块，需要手动安装，安装命令如下（CMD终端执行）：

pip install aiohttp lxml -i https://pypi.doubanio.com/simple/

3.2 重构核心逻辑：同步转异步

多线程脚本中，我们将单个章节的爬取任务拆分为crawl_chapter函数，用线程池分配任务；异步IO版本则需要将该函数改为异步函数，用async/await关键字标记，同时将同步的requests.get请求，改为异步的aiohttp.ClientSession.get请求。

核心优化点：

函数定义前添加async关键字，标记为异步函数；
用aiohttp.ClientSession替代requests，发送异步请求；
异步请求、获取响应文本时，添加await关键字，等待任务完成；
保留原有排版逻辑、文件写入逻辑，确保功能一致。

异步爬取函数代码：

# 异步爬取单个章节函数（核心优化部分）
async def crawl_chapter(session, i):
    try:
        # 提取章节链接，拼接完整URL（与多线程、单线程逻辑一致）
        a = i.xpath("./a/@href")[0]
        url = url_part + a

        # 异步发送请求（替换同步的requests.get）
        async with session.get(url, headers=headers) as resp:
            # 异步获取响应文本，指定编码格式，避免乱码
            html = await resp.text(encoding="utf-8")
            # 解析HTML（与多线程、单线程逻辑一致）
            tree = etree.HTML(html)

            # 提取章节标题（与原有逻辑一致）
            title = tree.xpath('//*[@id="neirong"]/h1/text()')
            chapter_title = title[0].strip() if title else "无标题章节"

            # 提取并优化正文排版（与原有逻辑一致）
            text_list = tree.xpath('//*[@id="txt"]//text()')
            # 过滤空文本，清理多余空格
            lines = [t.strip() for t in text_list if t.strip()]
            # 段落之间空一行，保持小说排版
            content = "\n\n".join(lines)
            full_content = f"\n{chapter_title}\n\n{content}\n\n"

            # 写入文件（与原有逻辑一致，文件写入是同步操作，无需异步）
            with open(f"novel/{chapter_title}.txt", "w", encoding="utf-8") as f:
                f.write(full_content)

            # 打印爬取进度
            print(f"已写入：{chapter_title}")
    except Exception as e:
        # 异常处理：单个章节失败，不影响其他章节
        print(f"爬取失败（章节异常）：{e}")

3.3 异步主函数：事件循环与任务调度

异步函数定义完成后，需要创建异步主函数，负责创建事件循环、管理异步会话、调度异步任务。核心步骤如下：

定义异步主函数main（添加async关键字）；
创建aiohttp.ClientSession对象（异步会话，用于发送所有异步请求）；
生成所有异步任务（将每个章节的爬取任务封装为asyncio任务）；
用asyncio.gather()方法，并发执行所有异步任务；
运行事件循环，执行异步主函数。

异步主函数代码：

# 异步主函数（管理事件循环和任务）
async def main():
    # 创建异步会话（复用会话，提升效率）
    async with aiohttp.ClientSession() as session:
        # 生成所有异步任务：将每个章节列表项传递给crawl_chapter函数
        tasks = [crawl_chapter(session, i) for i in chapter_list]
        # 并发执行所有异步任务，等待所有任务完成
        await asyncio.gather(*tasks)

这里需要注意：aiohttp.ClientSession建议复用，不要在每个异步任务中单独创建，否则会增加连接开销，影响爬取效率。asyncio.gather(*tasks)会并发执行所有任务，直到所有任务都完成，才会继续执行后续代码。

3.4 保留计时功能：对比三者耗时

为了直观看到异步IO的提速效果，我们保留上两篇中的计时逻辑，在脚本开始和结束时分别记录时间，最后输出总耗时，与单线程、多线程的耗时进行对比，让大家清晰看到异步IO的优势。

计时代码（位置与原有脚本一致）：

# 开始计时
start_time = time.time()

# 中间为核心爬取逻辑（异步主函数定义、任务调度等）

# 运行异步主函数，结束计时
asyncio.run(main())
end_time = time.time()
print(f"\n【异步IO版】总耗时：{end_time - start_time:.2f} 秒")

3.5 章节列表获取：同步即可，无需异步

需要特别说明：获取小说章节列表的操作，只需执行一次，耗时极短，无需异步处理，我们依然用同步的方式获取（这里可以继续使用requests库，也可以用aiohttp异步获取，差异不大）。本次实战中，我们沿用同步方式，保持代码简洁。

章节列表获取代码（与多线程、单线程一致）：

# 同步获取章节列表（只需一次，无需异步）
response = requests.get(url=url_html, headers=headers)
response.encoding = "utf-8"
tree = etree.HTML(response.text)
li = tree.xpath("/html/body/div[1]/div[4]/ul/li")
max_chapter = 10
chapter_list = li[:max_chapter]  # 只取前10章

四、完整可运行脚本（异步IO版 + 三者对比）

为了方便大家直接测试、对比效果，下面同时给出“单线程版、多线程版、异步IO版”三个完整脚本，大家可以分别运行，观察三者的耗时差异，直观感受异步IO的终极提速效果。

4.1 单线程版（衔接前文，无修改）

import requests
from lxml import etree
import os
import time

# 开始计时
start_time = time.time()

# 确保 novel 文件夹存在
if not os.path.exists("novel"):
    os.mkdir("novel")

url_part = "https://www.biquge365.net"
url_html = "https://www.biquge365.net/newbook/83621/"
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/138.0.0.0"
}
response = requests.get(url=url_html, headers=headers)
response.encoding = "utf-8"
tree = etree.HTML(response.text)
li = tree.xpath("/html/body/div[1]/div[4]/ul/li")
chapter_count = 0
max_chapter = 10

for i in li:
    if chapter_count >= max_chapter:
        break

    a = i.xpath("./a/@href")[0]
    url = url_part + a

    response = requests.get(url=url, headers=headers)
    response.encoding = "utf-8"
    tree = etree.HTML(response.text)

    title = tree.xpath('//*[@id="neirong"]/h1/text()')
    chapter_title = title[0].strip() if title else f"第{chapter_count + 1}章"

    text_list = tree.xpath('//*[@id="txt"]//text()')
    lines = [t.strip() for t in text_list if t.strip()]
    content = "\n\n".join(lines)
    full_content = f"\n{chapter_title}\n\n{content}\n\n"

    with open(f"novel/{chapter_title}.txt", "w", encoding="utf-8") as f:
        f.write(full_content)

    print(f"已写入：{chapter_title}")
    chapter_count += 1

# 结束计时
end_time = time.time()
print(f"\n【单线程】总耗时：{end_time - start_time:.2f} 秒")

4.2 多线程版（衔接前文，无修改）

import requests
from lxml import etree
import os
import time
from concurrent.futures import ThreadPoolExecutor

# 开始计时
start_time = time.time()

# 确保 novel 文件夹存在
if not os.path.exists("novel"):
    os.mkdir("novel")

url_part = "https://www.biquge365.net"
url_html = "https://www.biquge365.net/newbook/83621/"
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/138.0.0.0"
}

# 获取章节列表
response = requests.get(url=url_html, headers=headers)
response.encoding = "utf-8"
tree = etree.HTML(response.text)
li = tree.xpath("/html/body/div[1]/div[4]/ul/li")

max_chapter = 10
chapter_list = li[:max_chapter]

# 定义单个章节的爬取函数
def crawl_chapter(i):
    try:
        a = i.xpath("./a/@href")[0]
        url = url_part + a

        response = requests.get(url=url, headers=headers)
        response.encoding = "utf-8"
        tree = etree.HTML(response.text)

        title = tree.xpath('//*[@id="neirong"]/h1/text()')
        chapter_title = title[0].strip() if title else "无标题章节"

        text_list = tree.xpath('//*[@id="txt"]//text()')
        lines = [t.strip() for t in text_list if t.strip()]
        content = "\n\n".join(lines)
        full_content = f"\n{chapter_title}\n\n{content}\n\n"

        with open(f"novel/{chapter_title}.txt", "w", encoding="utf-8") as f:
            f.write(full_content)

        print(f"已写入：{chapter_title}")
    except Exception as e:
        print(f"爬取失败（章节异常）：{e}")

# 多线程执行
with ThreadPoolExecutor(max_workers=5) as executor:
    executor.map(crawl_chapter, chapter_list)

# 结束计时
end_time = time.time()
print(f"\n【多线程】总耗时：{end_time - start_time:.2f} 秒")

4.3 异步IO版（优化后，可直接运行）

import aiohttp
import asyncio
from lxml import etree
import os
import time
import requests  # 仅用于获取章节列表（同步，无需异步）

# 开始计时
start_time = time.time()

# 确保 novel 文件夹存在（与单线程、多线程逻辑一致）
if not os.path.exists("novel"):
    os.mkdir("novel")

# 基础配置（与单线程、多线程完全一致，无需修改）
url_part = "https://www.biquge365.net"
url_html = "https://www.biquge365.net/newbook/83621/"
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/138.0.0.0"
}

# 同步获取章节列表（只需一次，无需异步，与原有逻辑一致）
response = requests.get(url=url_html, headers=headers)
response.encoding = "utf-8"
tree = etree.HTML(response.text)
li = tree.xpath("/html/body/div[1]/div[4]/ul/li")

max_chapter = 10
# 只取前10章，控制爬取数量（与单线程、多线程一致）
chapter_list = li[:max_chapter]

# 异步爬取单个章节函数（核心优化部分）
async def crawl_chapter(session, i):
    try:
        a = i.xpath("./a/@href")[0]
        url = url_part + a

        # 异步发送请求，替代同步的requests.get
        async with session.get(url, headers=headers) as resp:
            # 异步获取响应文本，等待响应完成
            html = await resp.text(encoding="utf-8")
            tree = etree.HTML(html)

            title = tree.xpath('//*[@id="neirong"]/h1/text()')
            chapter_title = title[0].strip() if title else "无标题章节"

            text_list = tree.xpath('//*[@id="txt"]//text()')
            lines = [t.strip() for t in text_list if t.strip()]
            content = "\n\n".join(lines)
            full_content = f"\n{chapter_title}\n\n{content}\n\n"

            # 文件写入为同步操作，无需异步
            with open(f"novel/{chapter_title}.txt", "w", encoding="utf-8") as f:
                f.write(full_content)

            print(f"已写入：{chapter_title}")
    except Exception as e:
        print(f"爬取失败（章节异常）：{e}")

# 异步主函数（管理事件循环和任务调度）
async def main():
    # 复用异步会话，提升爬取效率
    async with aiohttp.ClientSession() as session:
        # 生成所有异步任务
        tasks = [crawl_chapter(session, i) for i in chapter_list]
        # 并发执行所有异步任务
        await asyncio.gather(*tasks)

# 运行异步主函数，启动事件循环
if __name__ == "__main__":
    asyncio.run(main())

    # 结束计时，输出总耗时
    end_time = time.time()
    print(f"\n【异步IO版】总耗时：{end_time - start_time:.2f} 秒")

五、真实测试结果对比与核心注意事项

5.1 三者耗时对比（真实测试）

我们在同一环境（Windows10、Python3.9、家用宽带、网络稳定）下，分别运行三个版本的脚本，爬取10章小说，测试结果如下（数据为多次测试平均值）：

异步IO版耗时：

从测试结果可以看出，异步IO版的耗时仅为单线程的1/15左右，比多线程还快2.3倍，提速效果非常显著。随着爬取章节数量的增加，这个差距会更大——比如爬取100章，单线程可能需要1到2分钟，多线程需要10到20秒，而异步IO仅需要5~8秒，效率优势一目了然。

提示：每个人的网络环境、电脑性能不同，耗时会有差异，但三者的速度排序始终是：异步IO > 多线程 > 单线程。

5.2 异步IO爬取核心注意事项

异步IO虽然速度快、资源占用低，但在使用过程中，有几个关键注意事项需要遵守，避免脚本报错、被网站反爬，确保爬取过程稳定。

区分同步与异步，避免混用：aiohttp是异步请求库，不能与requests（同步）混用在同一个异步任务中；文件写入、HTML解析等操作是同步的，无需添加await关键字，直接执行即可。
控制并发数，避免被反爬：异步IO的并发能力极强，默认情况下会同时发送所有请求，容易导致并发过高，被网站识别为爬虫，出现403、503等报错。如果爬取章节数量较多，可通过控制任务数量或添加延迟，限制并发速度。
必须添加异常处理：异步任务并发执行时，单个章节爬取失败（如链接失效、页面结构变化），如果没有异常处理，会导致整个事件循环崩溃，其他章节无法继续爬取。我们在crawl_chapter函数中添加了try-except异常处理，确保单个章节失败不影响整体任务。
复用异步会话：aiohttp.ClientSession建议复用，不要在每个异步任务中单独创建，否则会增加连接开销，降低爬取效率，甚至被网站识别为异常请求。
注意Python版本：asyncio.run()方法是Python3.7+才支持的，如果你的Python版本低于3.7，需要替换为手动创建事件循环（后续会讲解兼容方案）。
遵守网站规则，合规爬取：与单线程、多线程一样，异步IO爬虫仅用于学习和个人研究，请勿用于商业用途。爬取时请遵守网站的robots协议，不要过度频繁爬取，避免给网站服务器造成压力，否则可能会被封禁IP。

六、进阶思考与后续优化方向

本次异步IO优化，我们实现了爬虫速度的终极提升，但还有很多可优化的方向，后续博文会逐步讲解，帮助大家进一步完善脚本，提升爬取稳定性和效率：

添加延迟与并发控制：通过asyncio.sleep()添加请求延迟，或使用信号量（Semaphore）控制最大并发数，避免被网站反爬；
加入代理IP池：如果爬取数量过多，IP容易被封禁，可加入代理IP池，自动切换IP，提升爬取稳定性；
完善异常重试机制：对爬取失败的章节，添加自动重试逻辑，避免因网络波动导致的爬取不完整；
添加进度条显示：集成tqdm模块，实现异步任务爬取进度条，直观看到爬取进度；
批量合并章节：将所有章节合并为一个完整的txt文件，方便离线阅读，同时添加章节索引。

总结

本文衔接上一篇多线程爬虫博文，完成了异步IO（asyncio + aiohttp）的终极优化实战，核心亮点是“不改变原有功能、仅提升速度”，新手也能无缝衔接、直接使用。通过本次优化，我们不仅实现了爬虫速度比多线程再快2~4倍，还掌握了异步IO的核心原理和使用方法，理解了异步IO在IO密集型任务中的优势。

整个优化过程分为三步：替换同步请求为异步请求、定义异步函数、创建事件循环与任务调度，每一步都贴合实战，代码可直接复制运行。我们还对比了单线程、多线程、异步IO三种方式的耗时差异，让大家直观感受到异步IO的终极提速效果。

至此，我们已经完成了小说爬虫从单线程到多线程、再到异步IO的完整优化流程，从基础到进阶，逐步提升爬取效率。如果在运行脚本过程中遇到问题（比如异步报错、反爬、速度提升不明显），欢迎在评论区留言，我会一一解答。记得点赞、收藏、关注，后续会分享更多Python爬虫进阶技巧，帮助大家从新手成长为爬虫高手！