Python中如何同时运行多个Selenium实例

核心原则：每个调度触发的任务单独初始化专属WebDriver实例，任务全流程结束后强制释放所有关联资源，严格禁止实例跨任务复用，从根源上避免会话冲突、资源泄露问题

前置配置要求 #

• APScheduler必须替换默认单线程执行器，选用支持并行的执行器：
  • 优先选ThreadPoolExecutor，Selenium操作以IO等待为主，线程池资源开销远低于进程池，完全能满足并行需求
  • 执行器最大并发数按机器硬件配置设置，单Chrome无头实例常规占用200-500M内存，按可用内存推算实例上限即可，不要盲目开高并发导致内存溢出
• 每个Selenium实例必须做完全的上下文隔离：单独分配临时用户数据目录、随机分配调试端口，禁止复用默认配置路径，避免文件锁、端口冲突问题
• 任务配置要关闭同ID任务合并，放开同任务最大并行实例数限制，避免到触发时间时因为上一轮任务没跑完被调度器跳过

可直接复用的实现代码 #

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.executors.pool import ThreadPoolExecutor
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
import tempfile
import os
import shutil
import atexit

def scheduled_business_task(task_tag: str):
    driver = None
    temp_profile_dir = None
    try:
        # 为当前实例创建独立临时用户目录，彻底隔离配置
        temp_profile_dir = tempfile.mkdtemp(prefix=f"selenium_profile_{task_tag}_")
        chrome_options = Options()
        # 基础运行参数，可根据业务需求调整
        chrome_options.add_argument("--headless=new")
        chrome_options.add_argument("--no-sandbox")
        chrome_options.add_argument("--disable-dev-shm-usage")
        chrome_options.add_argument("--disable-gpu")
        chrome_options.add_argument(f"--user-data-dir={temp_profile_dir}")
        # 调试端口设为0，由系统自动分配可用端口，避免端口冲突
        chrome_options.add_argument("--remote-debugging-port=0")

        # 初始化当前任务专属WebDriver实例
        driver = webdriver.Chrome(options=chrome_options)
        driver.set_page_load_timeout(30)
        driver.implicitly_wait(5)

        # 以下替换为自身业务逻辑
        driver.get("https://example.com")
        print(f"任务[{task_tag}]执行完成，当前页面标题：{driver.title}")

    except Exception as e:
        print(f"任务[{task_tag}]运行异常：{repr(e)}")
    finally:
        # 无论任务成功/失败，必须彻底清理资源
        if driver:
            try:
                driver.quit()
            except Exception:
                pass
        # 清理临时用户目录
        if temp_profile_dir and os.path.exists(temp_profile_dir):
            try:
                shutil.rmtree(temp_profile_dir)
            except Exception:
                pass

if __name__ == "__main__":
    # 调度器配置：最大并发10个实例，按自身需求调整
    executors = {
        "default": ThreadPoolExecutor(max_workers=10)
    }
    job_defaults = {
        "coalesce": False,  # 关闭错过触发时间的任务合并
        "max_instances": 10 # 单个任务ID允许的最大并行实例数
    }
    scheduler = BlockingScheduler(executors=executors, job_defaults=job_defaults)

    # 测试：添加3个每10秒触发一次的任务，验证并行效果
    for idx in range(3):
        scheduler.add_job(
            scheduled_business_task,
            trigger="cron",
            args=[f"test_task_{idx}"],
            second="*/10"
        )

    # 注册全局退出钩子，服务终止时强制清理残留Chrome进程
    def kill_remaining_chrome():
        os.system("pkill -9 -f chrome")
    atexit.register(kill_remaining_chrome)

    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        pass

高频踩坑规避 #

• 禁止将WebDriver实例定义为全局变量、或在任务函数外初始化后传入任务，否则所有任务会复用同个实例，并行时会出现元素定位错乱、会话互相覆盖的问题
• 不要用driver.close()替代driver.quit()：close仅关闭当前标签页，不会终止浏览器主进程，长时间运行会堆积大量僵尸进程占满内存
• 高并发场景下提前调大系统文件句柄上限，避免出现too many open files报错
• 容器化部署时要给容器分配不小于1G的/dev/shm空间，默认64M空间会导致Chrome频繁崩溃
• 如果需要用Firefox、Edge等其他浏览器，隔离逻辑通用，替换对应WebDriver初始化代码即可

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