并行爬取URL并持续更新DataFrame/CSV价格列的最优方案

安全阿岳

2026-10-1

高效爬取方案理论说明

并行架构选型：爬取属于典型IO密集型任务，大部分时间消耗在网络请求等待响应上，因此优先选择多线程或协程架构，无需使用多进程：
- 多线程适配性最强，兼容requests、selenium等所有工具，Python GIL会在IO等待时自动释放，实际并行效率和多进程无差异，且线程创建/切换开销远低于进程
- 若所有页面都是静态渲染，可选协程（asyncio+aiohttp），并发性能比多线程更高，资源占用更低
循环调度逻辑优化：不要采用「整轮遍历全部URL→再从头开始下一轮」的模式，改为循环任务队列实现：
- 初始化时把所有URL和对应的DataFrame索引放入任务队列
- 每个爬取线程完成一个任务后，不管成功失败，都立刻把该任务重新放回队列尾部
- 优势：无需等待慢任务跑完就能启动下一轮爬取，资源利用率更高，各URL的爬取更新频率更均匀
数据更新安全设计：pandas DataFrame不是线程安全对象，直接多线程修改会导致数据错乱，最优实现是：
- 单独创建一个线程安全的结果队列，所有爬取线程爬取成功后，只把「行索引+最新价格」写入结果队列，不直接操作DataFrame
- 单独开1个常驻更新线程，唯一消费结果队列，按行索引更新DataFrame的价格列，全程无需加锁（只有一个线程写），性能最高且不会出现并发冲突
容错与限流配置：
- 每个爬取任务加超时、重试机制，失败次数超过阈值可标记后再放回队列，避免无效重复爬取
- 用信号量控制全局并发数，根据目标网站反爬强度设置10~50的并发量即可，避免被封IP
工具选择优先级：静态页面优先用requests+BeautifulSoup，性能最高；动态渲染页面优先用接口抓包，实在抓不到再用selenium，selenium建议用浏览器实例池复用，不要每个线程单独启动浏览器，开销过大

参考代码片段

import pandas as pd
import queue
import threading
from concurrent.futures import ThreadPoolExecutor
import requests
from bs4 import BeautifulSoup

# 初始化配置
MAX_WORKERS = 20  # 并发数，根据实际情况调整
RETRY_TIMES = 3  # 单任务最大重试次数

# 示例DataFrame，替换为你自己的df
df = pd.read_csv('your_data.csv')
# 初始化价格列如果不存在
if 'Price' not in df.columns:
    df['Price'] = None

# 任务队列：存储(行索引, URL, 失败次数)
task_queue = queue.Queue()
# 结果队列：存储(行索引, 最新价格)
result_queue = queue.Queue()

# 初始化任务队列
for idx, url in enumerate(df['URL']):
    task_queue.put((idx, url, 0))

def crawl_worker():
    while True:
        idx, url, retry_cnt = task_queue.get()
        try:
            # 爬取逻辑，根据不同网站自行修改
            resp = requests.get(url, timeout=10)
            resp.raise_for_status()
            soup = BeautifulSoup(resp.text, 'lxml')
            # 替换为实际的价格提取逻辑
            price = soup.select_one('.price').text.strip()
            # 爬取成功写入结果队列
            result_queue.put((idx, price))
            # 任务成功，放回队列重置失败次数
            task_queue.put((idx, url, 0))
        except Exception as e:
            # 失败重试，未超过阈值就放回队列
            if retry_cnt < RETRY_TIMES:
                task_queue.put((idx, url, retry_cnt + 1))
            else:
                # 超过重试次数，也可以标记价格为爬取失败
                task_queue.put((idx, url, 0))
        finally:
            task_queue.task_done()

def update_df_worker():
    """单独的df更新线程，唯一写df的逻辑"""
    while True:
        idx, price = result_queue.get()
        df.at[idx, 'Price'] = price
        # 如果需要实时落盘可以在这里加df.to_csv的逻辑，建议定期落盘不要每次都写
        result_queue.task_done()

if __name__ == '__main__':
    # 启动更新线程
    threading.Thread(target=update_df_worker, daemon=True).start()
    # 启动爬取线程池
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        for _ in range(MAX_WORKERS):
            executor.submit(crawl_worker)
    # 永久阻塞，任务队列会循环填充不会结束
    task_queue.join()

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