Python多进程加速网页爬取 快速写入Pandas DataFrame方法

现有ThreadPoolExecutor实现的常见问题 #

你写的并发版本能拿到1/max_workers的耗时提升，说明基础并发逻辑是通的，但这类写法普遍存在几个隐患，跑久了很容易出问题：

• 线程安全风险：直接在子线程里逐行写入全局pandas DataFrame是不安全的。pandas没有做多线程写入的锁保护，并发量上来之后很容易出现数据错位、列值覆盖、甚至DataFrame结构损坏，目前没出问题只是还没触发竞态条件。
• 连接复用缺失：如果每个请求还是单独调用requests.get()新建连接，会反复做TCP握手、TLS握手，平白多出来大量网络开销，根本吃不满并发带来的性能收益，并发开高了还容易触发目标站点的SYN洪水防护规则。
• 反爬适配不足：单纯堆高max_workers很容易触发亚马逊的反爬机制，一旦返回429状态码、验证码或者临时封IP，大量线程拿到的都是无效响应，反而比串行爬取效率更低；如果异常捕获逻辑没有做状态码区分，会把反爬拦截的响应直接判定为商品下架/属性缺失，直接污染价格追踪的数据集。
• 异常处理粗糙：网络波动、站点临时5xx错误、代理瞬时不可用这类可重试的异常，如果直接吞错跳过，会导致每日采集的数据缺失率居高不下；而商品404下架这类不可重试的请求，如果反复重试又会白白浪费请求配额。
• 写入逻辑低效：如果边爬边往DataFrame写数据，而不是先把所有爬取结果收集成结构化字典列表、最后一次性转成DataFrame，锁竞争和IO开销会额外拖慢整体速度。

可直接落地的优化方案 #

• 先改造请求底层：给每个线程绑定独立的requests.Session实例，挂载带重试策略的HTTP适配器，开启连接池复用，统一设置超时时间，不要让请求无限等待。参考实现：

import threading
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

thread_local = threading.local()

def init_session():
    session = requests.Session()
    # 重试策略：仅对5xx、429状态码做3次重试，退避间隔按指数增长
    retry_conf = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET"]
    )
    # 连接池大小和单线程并发能力匹配
    adapter = HTTPAdapter(max_retries=retry_conf, pool_connections=10, pool_maxsize=10)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    # 替换成真实浏览器UA，不要用requests默认的UA标识
    session.headers.update({"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"})
    return session

def get_thread_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = init_session()
    return thread_local.session

• 重构并发执行逻辑：不要在子线程操作全局DataFrame。把单个商品的详情页爬取、解析逻辑封装成独立函数，输入是单条商品的基础信息，输出是结构化的属性字典；主线程统一收集所有任务的返回结果，全部爬取完成后一次性生成最终DataFrame，从根源上规避多线程写入的安全问题。max_workers不要盲目开高，单出口IP下针对亚马逊这类站点设置8-15就足够，再高反爬拦截概率会陡增，几乎没有额外收益。核心逻辑参考：

from concurrent.futures import ThreadPoolExecutor, as_completed
import pandas as pd

def crawl_single_item(item: dict) -> dict:
    session = get_thread_session()
    base_info = {"item_id": item["item_id"], "item_name": item["item_name"], "detail_url": item["detail_url"]}
    try:
        resp = session.get(item["detail_url"], timeout=(3, 10))
        if resp.status_code == 404:
            return {**base_info, "is_on_sale": False, "crawl_status": "success"}
        # 这里写你的详情页属性解析逻辑
        # brand = 解析结果、category = 解析结果......
        return {
            **base_info,
            "is_on_sale": True,
            "current_price": "xxx",
            "brand": "xxx",
            "crawl_status": "success"
        }
    except Exception as e:
        # 记录失败原因，后续统一补爬，不要直接吞异常
        return {**base_info, "crawl_status": "failed", "error_msg": str(e)}

if __name__ == "__main__":
    # 第一阶段拿到的基础数据转成字典列表传入，不要直接传DataFrame行对象
    base_items = df_base.to_dict("records")
    crawl_results = []
    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(crawl_single_item, item) for item in base_items]
        for future in as_completed(futures):
            crawl_results.append(future.result())
    # 全量爬取完成后一次性生成最终DataFrame
    df_final = pd.DataFrame(crawl_results)

• 补全反爬适配：如果10个并发下还是频繁出现429、验证码拦截，不要硬加线程数，直接接入代理IP池做出口IP轮换，请求之间加0.5-2s的随机抖动延迟，不要用固定间隔发请求，很容易被反爬规则识别。如果遇到强JS渲染的验证场景，不要上Selenium，效率太低，可以换用Playwright无头模式，开启资源拦截屏蔽图片、CSS、非必要JS的加载，单实例爬取效率是Selenium的3-5倍。
• 做分层补爬机制：第一轮全量并发爬取完成后，把所有标记为爬取失败、被反爬拦截的商品单独捞出来，把并发数降到3-5做2-3轮补爬，整体采集成功率可以从70%左右提升到98%以上。
• 优化爬取策略：你是做价格历史追踪的，不需要每次全量爬10000个商品的详情页，可以给商品设置爬取优先级：近期价格波动大、参与促销的商品提高爬取频率，长期价格稳定、库存充足的商品降低爬取频率，每日需要爬的详情页请求量可以降到原来的30%以下。

进阶性能提升方向 #

• 如果你的详情页解析逻辑很重（比如需要处理大量正则匹配、复杂XPath提取），可以把IO密集的请求环节和CPU密集的解析环节拆分：用线程池并发拿响应文本，再把HTML内容丢给进程池做解析，规避Python GIL锁对CPU密集任务的影响，不过普通电商页面解析逻辑不复杂的话，这步收益不高。
• 如果担心爬取中途程序崩溃丢数据，可以每爬完100条就把当前收集到的结果追加写入Parquet/CSV文件，不要每次全量覆写DataFrame，IO开销会小很多。
• 后续爬取规模涨到10w+商品量级的时候，可以换用aiohttp+asyncio做异步请求，单进程并发效率比ThreadPoolExecutor高30%左右，不过1w商品的规模用优化后的线程池完全足够，10个并发下1.5小时以内就能跑完，完全满足每日爬取的需求。

内容的提问来源于stack exchange，提问作者Dandytomato