多线程Python爬虫：加速大规模学术文献采集

小白学大数据

发布于 2025-07-22 08:35:06

4470

1. 引言

在学术研究过程中，高效获取大量文献数据是许多科研工作者和数据分析师的需求。然而，传统的单线程爬虫在面对大规模数据采集时，往往效率低下，难以满足快速获取数据的要求。因此，利用多线程技术优化Python爬虫，可以显著提升数据采集速度，尤其适用于爬取学术数据库（如PubMed、IEEE Xplore、Springer等）。

2. 多线程爬虫的优势

2.1 单线程 vs. 多线程

单线程爬虫：顺序执行任务，一个请求完成后才发起下一个请求，导致I/O等待时间浪费。
多线程爬虫：并发执行多个请求，充分利用CPU和网络带宽，大幅提升爬取效率。

2.2 适用场景

需要快速爬取大量网页（如学术论文摘要、作者信息、引用数据等）。
目标网站允许一定程度的并发请求（需遵守**robots.txt**规则）。
数据采集任务可拆分为多个独立子任务（如分页爬取）。

3. 技术选型

技术	用途
<font style="color:rgb(64, 64, 64);background-color:rgb(236, 236, 236);">requests</font>	发送HTTP请求获取网页内容
<font style="color:rgb(64, 64, 64);background-color:rgb(236, 236, 236);">BeautifulSoup</font>	解析HTML，提取结构化数据
<font style="color:rgb(64, 64, 64);background-color:rgb(236, 236, 236);">concurrent.futures.ThreadPoolExecutor</font>	管理多线程任务
<font style="color:rgb(64, 64, 64);background-color:rgb(236, 236, 236);">fake_useragent</font>	随机生成User-Agent，避免反爬
<font style="color:rgb(64, 64, 64);background-color:rgb(236, 236, 236);">queue.Queue</font>	任务队列管理待爬取的URL
<font style="color:rgb(64, 64, 64);background-color:rgb(236, 236, 236);">csv</font> / <font style="color:rgb(64, 64, 64);background-color:rgb(236, 236, 236);">pandas</font>	存储爬取结果

4. 实现步骤

4.1 目标分析

假设我们需要从arXiv（开放学术论文库）爬取计算机科学领域的论文标题、作者、摘要和发布时间。arXiv的API允许批量查询，适合多线程爬取。

4.2 代码实现

（1）安装依赖

（2）定义爬虫核心函数

import requests
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
from concurrent.futures import ThreadPoolExecutor, as_completed
import pandas as pd
import time

# 设置随机User-Agent
ua = UserAgent()

# arXiv计算机科学分类的查询URL模板
ARXIV_URL = "https://arxiv.org/search/?query=cs&searchtype=all&start={}"

def fetch_page(start_index):
    """爬取单页数据"""
    url = ARXIV_URL.format(start_index)
    headers = {'User-Agent': ua.random}
    try:
        response = requests.get(url, headers=headers, timeout=10)
        if response.status_code == 200:
            soup = BeautifulSoup(response.text, 'html.parser')
            papers = []
            for paper in soup.select('.arxiv-result'):
                title = paper.select_one('.title').get_text(strip=True).replace('Title:', '')
                authors = paper.select_one('.authors').get_text(strip=True).replace('Authors:', '')
                abstract = paper.select_one('.abstract').get_text(strip=True).replace('Abstract:', '')
                published = paper.select_one('.is-size-7').get_text(strip=True)
                papers.append({
                    'title': title,
                    'authors': authors,
                    'abstract': abstract,
                    'published': published
                })
            return papers
    except Exception as e:
        print(f"Error fetching {url}: {e}")
    return []

def multi_thread_crawler(max_pages=100, workers=10):
    """多线程爬取"""
    results = []
    with ThreadPoolExecutor(max_workers=workers) as executor:
        futures = []
        for i in range(0, max_pages, 50):  # arXiv每页50条数据
            futures.append(executor.submit(fetch_page, i))
        
        for future in as_completed(futures):
            results.extend(future.result())
    
    return results

if __name__ == "__main__":
    start_time = time.time()
    papers = multi_thread_crawler(max_pages=200)  # 爬取200页（约10,000篇论文）
    df = pd.DataFrame(papers)
    df.to_csv("arxiv_papers.csv", index=False)
    print(f"爬取完成！耗时：{time.time() - start_time:.2f}秒，共获取{len(df)}篇论文。")

（3）代码解析

**fetch_page**：负责单页数据抓取，使用**BeautifulSoup**解析HTML并提取论文信息。
**multi_thread_crawler**：
- 使用**ThreadPoolExecutor**管理线程池，控制并发数（**workers=10**）。
- 通过**as_completed**监控任务完成情况，并合并结果。
数据存储：使用**pandas**将结果保存为CSV文件。

5. 优化与反爬策略

5.1 请求限速

避免被封IP，可在请求间增加延时：

import random
time.sleep(random.uniform(0.5, 2))  # 随机延时

5.2 代理IP

使用代理池防止IP被封：

proxyHost = "www.16yun.cn"
proxyPort = "5445"
proxyUser = "16QMSOML"
proxyPass = "280651"

proxies = {
    'http': f'http://{proxyUser}:{proxyPass}@{proxyHost}:{proxyPort}',
    'https': f'http://{proxyUser}:{proxyPass}@{proxyHost}:{proxyPort}'
}

response = requests.get(
    url, 
    headers=headers, 
    proxies=proxies,
    timeout=10
)

5.3 异常处理

增加重试机制：

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))
def fetch_page_with_retry(start_index):
    return fetch_page(start_index)