Python异步爬虫实战：用Aiohttp+BeautifulSoup 3秒抓取起点小说热销榜（附完整源码）

Python异步爬虫实战：3秒抓取起点小说热销榜的完整指南

最近在帮朋友分析网络文学市场趋势时，发现传统同步爬虫获取起点小说网榜单数据实在太慢。每次要等上几十秒才能拿到完整数据，效率低得让人抓狂。后来改用Aiohttp+BeautifulSoup的方案，竟然能在3秒内完成25页数据的抓取和存储。今天就把这个实战经验完整分享出来，包含你可能遇到的每个坑和解决方案。

1. 为什么选择异步爬虫？

去年在做电商价格监控系统时，我第一次体会到同步请求的局限性。当时要监控100个商品页面，用requests库串行请求需要近2分钟，而改用异步方式后仅需6秒。这个性能差距在小说榜单抓取中同样明显：

25页耗时

CPU占用率

代码复杂度

异步爬虫的核心优势在于：

• I/O密集型操作并行化

  ：当某个请求等待响应时，CPU可以处理其他请求

• 资源利用率高

  ：单线程就能实现高并发，避免多线程的切换开销

• 扩展性强

  ：轻松调整并发量应对不同规模的抓取需求

注意：异步编程适合I/O密集型场景，如果是计算密集型任务反而可能降低性能

2. 环境准备与工具选型

推荐使用Python 3.8+版本，这个版本区间在异步特性支持度和稳定性上达到最佳平衡。我的测试环境配置如下：

# 创建虚拟环境
python -m venv qidian_spider
source qidian_spider/bin/activate # Linux/Mac
qidian_spider\\Scripts\\activate # Windows

# 安装核心依赖
pip install aiohttp==3.8.4 beautifulsoup4==4.12.2 pandas==2.0.3

工具选型背后的考量：

• Aiohttp vs Requests

  ：前者支持异步，后者更简单易用

• BeautifulSoup vs lxml

  ：BS4容错性更好，适合不规范的HTML

• pandas

  ：简化数据存储和格式转换流程

常见安装问题解决方案：

3. 网页结构与反爬策略分析

起点小说网的热销榜页面(https://www.qidian.com/rank/hotsales)采用经典的分页设计，通过URL参数控制页码：

https://www.qidian.com/rank/hotsales/page1/
https://www.qidian.com/rank/hotsales/page2/
…

关键反爬措施及应对方案：

数据提取的关键CSS选择器：

• 书名：

  div.book-mid-info > h2

• 作者：

  a.name

• 简介：

  p.intro

• 更新日期：

  p.update

4. 完整实现代码解析

以下是经过生产环境验证的增强版代码，增加了异常处理和日志记录：

import aiohttp
import asyncio
from bs4 import BeautifulSoup
import csv
import pandas as pd
import logging
from datetime import datetime

# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s – %(levelname)s: %(message)s',
handlers=[logging.FileHandler('spider.log'), logging.StreamHandler()]
)

class QidianSpider:
def __init__(self, max_pages=25, concurrency=10):
self.base_url = "https://www.qidian.com/rank/hotsales/page{}/"
self.headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)'
}
self.max_pages = max_pages
self.semaphore = asyncio.Semaphore(concurrency)
self.data = []

async def fetch_page(self, session, page):
url = self.base_url.format(page)
try:
async with self.semaphore:
async with session.get(url, headers=self.headers) as response:
if response.status != 200:
logging.error(f"Page {page} failed: {response.status}")
return None
return await response.text()
except Exception as e:
logging.error(f"Error fetching page {page}: {str(e)}")
return None

def parse_page(self, html, page):
if not html:
return

soup = BeautifulSoup(html, 'html.parser')
books = soup.select('div.book-mid-info')

for book in books:
try:
item = {
'title': book.h2.get_text(strip=True),
'author': book.select_one('a.name').get_text(strip=True),
'intro': book.select_one('p.intro').get_text(strip=True),
'update': book.select_one('p.update').get_text(strip=True),
'page': page
}
self.data.append(item)
except AttributeError as e:
logging.warning(f"Parse error on page {page}: {str(e)}")

async def process_page(self, session, page):
start = datetime.now()
html = await self.fetch_page(session, page)
if html:
self.parse_page(html, page)
elapsed = (datetime.now() – start).total_seconds()
logging.info(f"Page {page} processed in {elapsed:.2f}s")

async def run(self):
async with aiohttp.ClientSession() as session:
tasks = [self.process_page(session, p) for p in range(1, self.max_pages+1)]
await asyncio.gather(*tasks)

self.save_results()

def save_results(self):
# CSV存储
with open('qidian_rank.csv', 'w', encoding='utf-8-sig', newline='') as f:
writer = csv.DictWriter(f, fieldnames=['title','author','intro','update','page'])
writer.writeheader()
writer.writerows(self.data)

# Excel格式转换
pd.read_csv('qidian_rank.csv').to_excel('qidian_rank.xlsx', index=False)
logging.info(f"Saved {len(self.data)} books to qidian_rank.xlsx")

if __name__ == '__main__':
start_time = datetime.now()
spider = QidianSpider(max_pages=25)

loop = asyncio.get_event_loop()
loop.run_until_complete(spider.run())

total_time = (datetime.now() – start_time).total_seconds()
logging.info(f"Total execution time: {total_time:.2f} seconds")

代码优化点说明：

5. 性能优化技巧

通过多次实测，总结出以下提速方案：

连接池配置

conn = aiohttp.TCPConnector(
limit=20, # 最大连接数
force_close=True,
enable_cleanup_closed=True
)
async with aiohttp.ClientSession(connector=conn) as session:
# 你的代码

超时设置

timeout = aiohttp.ClientTimeout(total=10, sock_connect=3)
async with session.get(url, timeout=timeout) as response:
# 你的代码

缓存策略

对不变的数据可以考虑使用：

from cachetools import TTLCache
cache = TTLCache(maxsize=100, ttl=3600)

async def get_with_cache(url):
if url in cache:
return cache[url]
content = await fetch(url)
cache[url] = content
return content

常见性能瓶颈排查表

可能原因

解决方案

6. 数据存储与后续处理

原始数据存储为CSV后，可以用pandas进行深度分析：

df = pd.read_csv('qidian_rank.csv')

# 分析更新频率
df['update_time'] = df['update'].str.extract(r'(\\d{4}-\\d{2}-\\d{2})')
update_counts = df['update_time'].value_counts().sort_index()

# 作者作品数量统计
author_stats = df['author'].value_counts().head(10)

# 保存分析结果
with pd.ExcelWriter('analysis.xlsx') as writer:
update_counts.to_excel(writer, sheet_name='更新频率')
author_stats.to_excel(writer, sheet_name='高产作者')

对于大规模数据，建议使用：

• MongoDB

  ：适合非结构化数据存储

• SQLite

  ：轻量级关系型数据库

• Elasticsearch

  ：支持全文检索和分析

7. 项目扩展方向

这个基础框架可以轻松扩展更多功能：

在最近的一次实际应用中，我用这个框架同时监控了三个小说网站的榜单变化，配合自动邮件报告功能，每天都能第一时间获取市场动态。异步爬虫带来的效率提升，让原本需要专人值守的工作变成了全自动流程。