同URL多实例Python Selenium表单提交并行爬取方案咨询

核心疑问解答 #

• 你对I/O bound场景的判断有误。I/O密集型的判定核心是任务总耗时中等待I/O响应的占比，和最终爬取到的数据量大小没有关系。你提交表单后等待服务器响应、页面渲染的时间都属于I/O等待时间，所以你的场景仍然是典型的I/O密集型，不属于常规Web scraper的例外情况。
• Chrome默认启用多进程架构，会为渲染引擎、标签页、扩展、GPU进程等分别启动独立子进程，所以你跑单线程脚本时看到的多核心CPU占用是Chrome本身的进程消耗，和Python侧的脚本逻辑无关。
• Python侧多进程的核心优势是突破GIL锁限制利用多核CPU，仅当你的Python脚本本身存在大量CPU密集的预处理/后解析逻辑时才有明显收益。你当前场景下Python侧仅负责给ChromeDriver发指令、拿结果，逻辑极轻，用多进程的最大作用其实是隔离不同Chrome实例的驱动资源，避免多线程下可能出现的Driver冲突，但性能收益和多线程差异很小。
• 同时运行大量Chrome实例确实会让整体任务变成CPU密集型。每个Chrome实例本身就会占用至少几百MB内存和一定CPU资源跑渲染、JS执行逻辑，当并发实例数超过CPU承载上限后，Chrome本身的资源争抢会成为瓶颈，此时再提升Python侧的并发数不会带来性能提升，反而会让整体耗时变长。

架构设计建议 #

• 优先控制并发数：初期建议同时运行的Chrome实例数不超过CPU核心数的1~2倍，后续可以根据实际运行的资源占用情况调整，避免出现CPU占满导致的任务卡顿。
• 优先选择多线程实现：多线程的内存开销、启动开销远低于多进程，且完全适配你当前I/O密集的场景，实现更简单，没有进程间通信的额外成本。
• 高频任务可选Driver池优化：如果需要跑的任务量很大，可以复用已启动的Chrome实例，每次跑完任务清理Cookie、本地存储后再接下一个任务，减少Chrome启停阶段的CPU峰值开销和启动等待时间。
• 仅当Python侧有大量CPU计算逻辑时，再切换为多进程实现，将计算逻辑和Selenium调用解耦。

实现伪代码参考 #

以下是多线程版本的实现，若要切换为多进程仅需把ThreadPoolExecutor替换为ProcessPoolExecutor即可：

from concurrent.futures import ThreadPoolExecutor, as_completed
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager

# 固定目标表单页URL
TARGET_URL = "你的目标表单页面地址"
# 待提交的不同表单数据列表
form_data_list = [
    {"query": "测试关键词1", "other_field": "值1"},
    {"query": "测试关键词2", "other_field": "值2"},
    # 更多待提交数据
]

def scrape_single_task(form_data):
    # 每个线程单独初始化Chrome实例，避免资源冲突
    driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()))
    try:
        driver.get(TARGET_URL)
        # 填入表单逻辑示例，可根据实际页面调整
        driver.find_element("id", "input_box").send_keys(form_data["query"])
        driver.find_element("id", "submit_btn").click()
        # 显式等待结果加载，可根据实际情况调整等待条件
        driver.implicitly_wait(10)
        # 抓取目标结果
        result_text = driver.find_element("id", "result_area").text
        return {"data": form_data, "result": result_text, "status": "success"}
    except Exception as e:
        return {"data": form_data, "error": str(e), "status": "failed"}
    finally:
        # 强制关闭实例，避免残留进程占资源
        driver.quit()

if __name__ == "__main__":
    # 并发数，根据机器性能调整，初期建议设为4~8
    MAX_CONCURRENCY = 4
    final_results = []

    with ThreadPoolExecutor(max_workers=MAX_CONCURRENCY) as executor:
        task_futures = [executor.submit(scrape_single_task, data) for data in form_data_list]
        for future in as_completed(task_futures):
            final_results.append(future.result())
    
    # 后续处理所有爬取结果
    print(final_results)

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