关于Python环境下多进程加速Selenium爬虫及大规模URL爬取优化的技术咨询

嗨，针对你爬取120万URL遇到的吞吐量低、内存暴涨问题，我来一步步给你拆解可行的优化方案：

一、多进程是不是这类任务的最佳选择？ #

答案是否定的，你当前的代码存在一个核心问题：每次调用scrape函数都会新建一个Chrome Driver实例，每个Chrome实例本身就要占用几百MB内存，4个进程已经是资源的极大消耗，再扩进程数只会快速耗尽系统内存——这也是你内存持续上涨的主要原因。

多进程更适合CPU密集型任务，但你的爬虫是IO密集型（大部分时间在等网页加载），而且Selenium的Chrome实例属于重量级资源，多进程的资源隔离特性反而会导致资源利用率极低，完全发挥不出多进程的优势。

二、要不要切换到asyncio + Playwright？ #

非常推荐你切换！ 这会是解决你问题的关键一步：

• Playwright天生支持异步(asyncio)，可以在单个浏览器实例中开启多个上下文/标签页，资源占用比Selenium+多进程低一个量级；
• Playwright的API设计更简洁，内置了自动等待元素加载的机制，减少了手动处理超时的麻烦；
• 异步IO模型在IO密集型的爬虫任务中，吞吐量远高于多进程，同时内存控制更出色——你可以用更少的系统资源支撑更多的并发请求。

三、如何控制爬取速率、实现智能批处理？ #

1. 速率与并发控制 #

• 用信号量(Semaphore)限制并发数：不管是asyncio还是多进程，都可以用信号量来控制同时进行的请求数，比如限制最多15个并发，避免触发目标网站的反爬机制，也防止系统资源过载；
• 添加随机延迟：在请求之间加入少量随机sleep（比如1-3秒），模拟人类访问行为，降低被封IP的风险；
• 结合代理池：你提到的旋转代理要和速率控制配合，每个代理的请求频率不要过高，避免代理被目标网站封禁。

2. 智能批处理 #

• 分批加载URL：不要一次性把120万URL都加载到内存里，建议从文件（比如CSV、TXT）中分批读取，每次读取1000-5000个URL，爬完一批就将结果写入文件，再加载下一批，避免内存被大量URL或爬取结果占满；
• 批量保存结果：不要把所有爬取结果都存在results列表中，每批爬完就用JSON Lines格式（每行一个JSON对象）写入文件，这样既节省内存，也能避免程序崩溃时丢失所有数据。

优化后的代码示例 #

方案1：asyncio + Playwright（推荐） #

import asyncio
import json
from playwright.async_api import async_playwright
from bs4 import BeautifulSoup

async def scrape(url, semaphore):
    async with semaphore:
        try:
            async with async_playwright() as p:
                browser = await p.chromium.launch(headless=True)
                context = await browser.new_context()
                page = await context.new_page()
                await page.goto(url, timeout=30000)
                html = await page.content()
                await browser.close()
                
                soup = BeautifulSoup(html, "lxml")
                title = soup.find("title").get_text(strip=True)
                return {"url": url, "title": title}
        except Exception as e:
            return {"url": url, "error": str(e)}

async def main():
    # 限制并发数为10，可根据系统资源调整
    semaphore = asyncio.Semaphore(10)
    urls = [
        "https://pitchbook.com/profiles/company/168089-41",
        "https://pitchbook.com/profiles/company/111349-63",
        "https://pitchbook.com/profiles/company/121201-84",
        # ... 其他URL
    ]
    # 每批处理500个URL，可根据内存情况调整
    batch_size = 500
    for i in range(0, len(urls), batch_size):
        batch_urls = urls[i:i+batch_size]
        tasks = [scrape(url, semaphore) for url in batch_urls]
        results = await asyncio.gather(*tasks)
        # 追加写入结果文件
        with open("scrape_results.jsonl", "a", encoding="utf-8") as f:
            for res in results:
                f.write(f"{json.dumps(res, ensure_ascii=False)}\n")

if __name__ == "__main__":
    asyncio.run(main())

方案2：优化现有Selenium+多进程代码（不换栈的情况下） #

核心是复用每个进程的Chrome Driver实例，避免每次请求都新建：

import json
from multiprocessing import Pool
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from bs4 import BeautifulSoup

# 每个进程只初始化一次driver，复用资源
driver = None
def init_driver():
    global driver
    options = Options()
    options.add_argument("--headless=new")
    options.add_argument("--disable-gpu")
    driver = webdriver.Chrome(options=options)

def scrape(url):
    try:
        driver.get(url)
        html = driver.page_source
        soup = BeautifulSoup(html, "lxml")
        title = soup.find("title").get_text(strip=True)
        return {"url": url, "title": title}
    except Exception as e:
        return {"url": url, "error": str(e)}

if __name__ == "__main__":
    urls = [
        "https://pitchbook.com/profiles/company/168089-41",
        "https://pitchbook.com/profiles/company/111349-63",
        "https://pitchbook.com/profiles/company/121201-84",
        # ... 其他URL
    ]
    # 用initializer初始化每个进程的driver
    with Pool(processes=4, initializer=init_driver) as pool:
        batch_size = 500
        for i in range(0, len(urls), batch_size):
            batch_urls = urls[i:i+batch_size]
            results = pool.map(scrape, batch_urls)
            # 分批写入结果，避免内存溢出
            with open("selenium_results.jsonl", "a", encoding="utf-8") as f:
                for res in results:
                    f.write(f"{json.dumps(res, ensure_ascii=False)}\n")

额外的反爬建议 #

• 除了旋转代理，还可以随机切换User-Agent，Playwright和Selenium都支持设置自定义User-Agent；
• 避免固定的请求间隔，用随机范围的延迟（比如random.uniform(1,3)）；
• 如果目标网站有登录限制，可以在Playwright/Selenium中复用登录后的上下文，避免每次请求都重新登录。

内容来源于stack exchange