当前位置：首页 > news >正文

Python协程实战：异步高效爬取《鬼神传》全本小说

news 2026/5/31 20:43:34

一、项目背景

在网络爬虫开发中，同步爬取大量小说章节时效率低下，等待IO时间过长。本文将使用Python协程和异步IO技术，结合 aiohttp 、 asyncio 、 aiofiles 实现高并发、高效率的小说爬取，大幅提升下载速度。

二、技术选型

异步HTTP请求：aiohttp

异步文件写入：aiofiles

HTML解析：lxml

协程调度：asyncio

网页请求：requests

三、完整代码实现

import requests from lxml import etree import time import asyncio import aiohttp import aiofiles import os BASE_URL = "https://www.zanghaihua.org" # 获取所有章节链接 def get_every_chapter_url(url): headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" } resp = requests.get(url, headers=headers) resp.encoding = "utf-8" tree = etree.HTML(resp.text) a_list = tree.xpath('//dl[@class="gs-booklist-dl"]//dd/a') href_list = [] title_list = [] for a in a_list: href = a.xpath('./@href')[0] title = a.xpath('./text()')[0] full_url = BASE_URL + href href_list.append(full_url) title_list.append(title) print(f"成功获取 {len(href_list)} 章目录") return href_list, title_list # 下载单章 async def download_one(session, url, title): headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Accept-Encoding": "gzip, deflate" } try: async with session.get(url, headers=headers, timeout=20) as resp: page_text = await resp.text(encoding="utf-8", errors="ignore") tree = etree.HTML(page_text) content_list = tree.xpath('//div[@class="gs-article-text"]//p//text()') content = "\n".join([text.strip() for text in content_list if text.strip()]) if not os.path.exists("./鬼神传"): os.mkdir("./鬼神传") async with aiofiles.open(f"./鬼神传/{title}.txt", "w", encoding="utf-8") as f: await f.write(title + "\n\n" + content) print(f"已保存：{title}") except Exception as e: print(f"下载失败 {title}：{e}") # 批量下载 async def download(href_list, title_list): async with aiohttp.ClientSession() as session: tasks = [] for url, title in zip(href_list, title_list): task = asyncio.create_task(download_one(session, url, title)) tasks.append(task) await asyncio.gather(*tasks) # 主函数 def main(): start = time.time() book_url = "https://www.zanghaihua.org/guwen/guishenchuan/" href_list, title_list = get_every_chapter_url(book_url) asyncio.run(download(href_list, title_list)) end = time.time() print(f"《鬼神传》全部下载完成！总耗时：{end - start:.2f} 秒") if __name__ == "__main__": main()

四、代码详解

1. 目录获取函数（同步）

def get_every_chapter_url(url): # 请求头伪装浏览器 headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" } resp = requests.get(url, headers=headers) resp.encoding = "utf-8" tree = etree.HTML(resp.text) # XPath提取所有章节链接与标题 a_list = tree.xpath('//dl[@class="gs-booklist-dl"]//dd/a')

用 requests 同步获取目录页（目录页仅1次请求，同步无性能影响）

XPath 精准定位章节 <a> 标签，提取 href 与 title

2. 单章异步下载

async def download_one(session, url, title): async with session.get(url, headers=headers, timeout=20) as resp: page_text = await resp.text(encoding="utf-8", errors="ignore") tree = etree.HTML(page_text) content_list = tree.xpath('//div[@class="gs-article-text"]//p//text()')

async/await 实现异步非阻塞请求

session.get 复用连接，提升效率

异常捕获保证单个章节失败不影响整体

3. 异步批量下载

async def download(href_list, title_list): async with aiohttp.ClientSession() as session: tasks = [asyncio.create_task(download_one(session, url, title)) for url, title in zip(href_list, title_list)] await asyncio.gather(*tasks)

创建任务列表， gather 并发执行

真正实现多章节同时下载，速度提升10~50倍

4. 主函数调度

def main(): start = time.time() href_list, title_list = get_every_chapter_url(book_url) asyncio.run(download(href_list, title_list)) end = time.time() print(f"《鬼神传》全部下载完成！总耗时：{end - start:.2f} 秒")

计时统计，直观展示异步爬取效率

五、运行效果

几十章内容10秒内完成，同步爬取需30秒以上

六、关键优化点

1. 连接复用： aiohttp.ClientSession 减少TCP握手开销

2. 异步文件： aiofiles 避免磁盘IO阻塞事件循环

3. 异常处理：单章下载失败不中断整体任务

4. 编码兼容： errors="ignore" 避免乱码崩溃

5. 目录自动创建：自动生成 ./鬼神传文件夹