如何使用Selenium WebDriver正确处理多URL调用并避免线程问题？

你的核心问题在于全局单例的WebDriver实例在多线程环境下会被共享，导致线程间操作冲突。要实现每个URL对应独立的驱动实例（或每个线程拥有独立实例），可以采用以下两种方案：

方案1：单线程循环中为每个URL创建独立驱动 #

适合单线程批量处理URL的场景，每个URL处理完成后立即销毁驱动，完全隔离实例，从根源避免线程冲突：

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
from bs4 import BeautifulSoup
import constants

def create_new_driver():
    # 每次调用都创建全新的驱动实例
    options = Options()
    options.add_argument('--headless')
    return webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=options)

def clickQuarterlyButton(url):
    driver = create_new_driver()
    try:
        driver.get(url)
        WebDriverWait(driver, constants.SLEEP_TIME).until(
            EC.element_to_be_clickable((By.XPATH, '//*[@id="Col1-1-Financials-Proxy"]/section/div[1]/div[2]/button'))
        ).click()
        soup = BeautifulSoup(driver.page_source, 'html.parser')
        return soup
    finally:
        # 无论是否出错，都确保驱动被正确关闭
        driver.close()
        driver.quit()

# 循环处理URL
for url in urls:
    clickQuarterlyButton(url)

方案2：多线程环境下用线程本地存储隔离实例 #

如果用多线程批量处理URL，希望减少驱动创建销毁的开销，同时避免线程冲突，可以用threading.local为每个线程分配独立的驱动实例：

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
from bs4 import BeautifulSoup
import constants
import threading
from concurrent.futures import ThreadPoolExecutor

# 线程本地存储：每个线程拥有独立的driver实例，互不干扰
thread_local = threading.local()

def get_thread_driver():
    # 仅为当前线程创建一次驱动实例
    if not hasattr(thread_local, 'driver'):
        options = Options()
        options.add_argument('--headless')
        thread_local.driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=options)
    return thread_local.driver

def quit_thread_driver():
    # 关闭当前线程的驱动实例
    if hasattr(thread_local, 'driver'):
        thread_local.driver.close()
        thread_local.driver.quit()
        del thread_local.driver

def clickQuarterlyButton(url):
    driver = get_thread_driver()
    driver.get(url)
    WebDriverWait(driver, constants.SLEEP_TIME).until(
        EC.element_to_be_clickable((By.XPATH, '//*[@id="Col1-1-Financials-Proxy"]/section/div[1]/div[2]/button'))
    ).click()
    soup = BeautifulSoup(driver.page_source, 'html.parser')
    return soup

# 多线程处理URL示例
def process_single_url(url):
    try:
        return clickQuarterlyButton(url)
    finally:
        quit_thread_driver()

with ThreadPoolExecutor(max_workers=5) as executor:
    executor.map(process_single_url, urls)

核心要点 #

1. 原代码的问题根源是全局变量global_driver被所有线程共享，多个线程同时读写会导致驱动实例混乱、操作冲突。
2. 方案1完全隔离每个URL的驱动实例，适合单线程场景，缺点是频繁创建销毁驱动会增加性能开销。
3. 方案2通过线程本地存储实现线程级别的实例隔离，既避免了线程冲突，又减少了驱动创建销毁的次数，适合多线程批量处理场景。
4. 两种方案都用finally块保证驱动在使用后被正确关闭，避免资源泄漏。

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