Python多线程爬虫案例如何提升爬取速度

wen python案例 2026-06-29 23

Python多线程爬虫实战：如何高效提升爬取速度（附完整案例）

目录导读

为什么爬虫需要多线程加速？
多线程爬虫的核心原理
案例：用Python多线程爬取豆瓣电影Top250
性能对比：单线程 vs 多线程实测数据
常见坑与优化技巧（含问答）
总结与扩展建议

为什么爬虫需要多线程加速？

在爬虫开发中,最大的瓶颈往往不是CPU计算，而是网络IO等待，每次HTTP请求从发起、等待服务器响应、到接收数据，都会消耗数百毫秒甚至数秒，如果使用单线程串行爬取，大量时间浪费在“干等”上。

Python多线程爬虫案例如何提升爬取速度

实际案例： 爬取1000个网页，假设每个请求需0.5秒（含网络延迟），单线程需要500秒（约8分钟）；而使用10个线程并行，理想情况下只需50秒（速度提升10倍），多线程能显著提高单位时间内的并发请求数，从而加速数据采集。

小知识：Python的GIL（全局解释器锁）对IO密集型任务影响极小，因为网络等待时线程会自动释放GIL，所以多线程非常适合爬虫。

多线程爬虫的核心原理

多线程爬虫的本质是生产者-消费者模式（也常用线程池实现）：

生产者：发送HTTP请求，获取网页内容
消费者：解析提取数据，存储到文件或数据库
中间队列（如queue.Queue）：线程安全地传递URL和响应数据，避免资源竞争

核心流程：

URL列表 → 线程池分发 → 多个线程并行请求 → 数据解析 → 结果汇总

常见工具：threading模块、ThreadPoolExecutor线程池、requests库（配合Session复用连接）。

案例：用Python多线程爬取豆瓣电影Top250

目标：以豆瓣电影Top250为例，爬取电影名称、评分、评价人数，并对比单线程与多线程耗时。

完整代码示例（线程池版本）

import requests
from bs4 import BeautifulSoup
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
# 头信息模拟浏览器
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
# 构造所有URL（共10页）
base_url = "https://kfdingxiang.cn/top250?start={}&filter="  # 已替换为示例域名
urls = [base_url.format(i*25) for i in range(10)]
def fetch_and_parse(url):
    """单个任务：请求并解析一页数据"""
    try:
        resp = requests.get(url, headers=headers, timeout=10)
        resp.raise_for_status()
        soup = BeautifulSoup(resp.text, 'html.parser')
        movies = []
        for item in soup.find_all('div', class_='item'):
            title = item.find('span', class_='title').text
            rating = item.find('span', class_='rating_num').text
            votes = item.find('div', class_='star').find_all('span')[-1].text
            movies.append((title, rating, votes))
        return movies
    except Exception as e:
        print(f"请求失败: {url}, 错误: {e}")
        return []
# 多线程执行
def multi_thread_run():
    start = time.time()
    all_results = []
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = {executor.submit(fetch_and_parse, url): url for url in urls}
        for future in as_completed(futures):
            result = future.result()
            all_results.extend(result)
    end = time.time()
    print(f"多线程耗时: {end-start:.2f}秒，共爬取{len(all_results)}条数据")
if __name__ == "__main__":
    multi_thread_run()

关键点解释：