Python读取URL列表实现多线程网页爬取与结果合并方案

关于是否需要手动按线程数拆分输入文件 #

完全不需要手动拆分。手动切分任务文件不仅增加冗余代码，还很容易出现负载不均——比如部分线程分到的URL响应快、早早跑完空等，部分线程分到的URL响应慢、长时间积压，整体资源利用率上不去。
直接使用Python标准库concurrent.futures提供的ThreadPoolExecutor即可，组件内部自带任务队列，会自动将待处理任务分配给空闲线程，全程不需要人工干预任务拆分逻辑。

核心实现逻辑 #

改造原则 #

• 所有本地预处理操作（读取输入JSON、URL去重、过滤空链接）放在主线程执行，这部分是纯本地计算、IO开销极低，不会影响整体效率。
• 仅把网络请求、页面解析这类IO密集型操作放到线程池执行，每个线程独立处理单条URL任务，返回单条爬取结果，不在线程间共享可变数据，从根源上避免线程安全问题，不需要加锁。
• 所有任务执行完成后，在主线程统一完成结果合并、输出JSON写入操作，和原有串行逻辑的输出完全兼容。
• 顺手修复原代码的隐藏bug：原代码中如果触发AttributeError找不到feed链接时，feed_details_url变量未定义，后续判断会抛出UnboundLocalError，改造时提前初始化变量即可。

线程数配置建议 #

爬取任务属于IO密集型，线程数不需要设置过高，一般配置为CPU核心数 * 5 ~ 10即可（比如8核CPU配置40~80线程）。线程数过高反而容易触发目标网站限流、被封IP，额外增加网络错误率。
额外建议给requests.get加上超时参数，避免偶发的卡死请求挂住线程；可以增加简单的重试逻辑应对临时网络波动。

改造后完整代码 #

import contextlib
from concurrent.futures import ThreadPoolExecutor, as_completed
from bs4 import BeautifulSoup
import feedparser
import pandas
import requests
import time

BASE_URL = 'https://www.iso.org'
# 线程数可根据自己的网络环境、目标站点限流规则调整
MAX_WORKERS = 50
REQUEST_TIMEOUT = 10

def create_iso_details_json(p_merged_iso_df):
    merged_iso_details_df = p_merged_iso_df.drop_duplicates(subset=['Link']).drop(columns=['TC', 'ICS'], axis=1)
    # 过滤空链接，生成待处理任务列表
    task_list = [
        (iso, stage, link)
        for iso, stage, link in zip(
            merged_iso_details_df['Standard and/or project'],
            merged_iso_details_df['Stage'],
            merged_iso_details_df['Link']
        )
        if link != ''
    ]

    iso_details_dfs = []
    # 启动线程池处理任务
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        # 提交所有任务
        future_map = {
            executor.submit(parse_iso_details, iso, stage, link): (iso, link)
            for iso, stage, link in task_list
        }
        # 逐批收集完成的任务结果
        for future in as_completed(future_map):
            res = future.result()
            if res is not None:
                iso_details_dfs.append(res)

    merged_iso_details_df = pandas.concat(iso_details_dfs)
    print('Total rows retrieved: ', len(merged_iso_details_df.index))
    merged_iso_details_df.to_json('iso_details.json', orient="records")
    
def parse_iso_details(p_iso, p_stage, p_url):
    print('Processing URL: ', p_url)
    # 提前初始化变量，修复原代码未定义bug
    feed_details_url = None
    try:
        resp = requests.get(p_url, timeout=REQUEST_TIMEOUT)
        resp.raise_for_status()
        soup = BeautifulSoup(resp.text, 'html.parser')
        feed_link = soup.find('section', {'id': 'product-details'}).find('a', {'class': 'ss-icon ss-social-circle text-warning text-sm'})
        if feed_link:
            feed_details_url = BASE_URL + feed_link['href']
    except (AttributeError, requests.exceptions.RequestException) as e:
        print(f'Could not get valid data for URL {p_url}, error: {str(e)}')
        return None

    if not feed_details_url:
        return None
    
    print('Found feed URL: ', feed_details_url)
    try:
        return read_iso_details(feed_details_url, p_iso, p_stage)
    except Exception as e:
        print(f'Parse feed failed for URL {feed_details_url}, error: {str(e)}')
        return None
    
def read_iso_details(p_feed_details_url, p_iso, p_stage):
    data = {'Standard and/or project': p_iso, 'Stage': p_stage}
    df = pandas.DataFrame(data, index=[0])
    feed = feedparser.parse(p_feed_details_url)
    df['Publication date'] = [entry.published for entry in feed.entries]
    return df

def main():
    start_time = time.time()
    merged_iso_df = pandas.read_json('input_file.json', dtype={"Stage": str})
    create_iso_details_json(merged_iso_df)
    print(f"--- {time.time() - start_time} seconds ---")

if __name__ == "__main__":
    main()

结果合并说明 #

上述实现不需要做复杂的跨线程数据同步：

1. 每个线程独立处理单条任务，返回单条结果对应的小DataFrame，处理失败直接返回None
2. 主线程通过as_completed迭代器收集所有已经完成的任务结果，过滤掉None值后存入列表
3. 所有任务跑完后，直接调用pandas.concat()一次性合并所有结果，和原代码的合并逻辑完全一致，最终输出的JSON格式和串行版本没有区别。

这种实现方式比手动拆分文件、多线程共享列表写数据的方案代码更简洁，没有线程安全隐患，负载分配更均匀，在网络正常、目标站点不限流的情况下，整体耗时可以降到原来串行版本的1/20~1/50。

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