如何在Python Selenium中实现多线程：为多ChromeDriver分配唯一文件数据

核心思路：用线程安全队列分配任务，每个线程独立管理ChromeDriver，通过线程锁保证结果写入单个文件的安全性，既避免任务重复，又不用事后合并多文件。

关键实现要点 #

• 任务队列：把Excel中所有待处理数据提前存入线程安全队列，线程从队列取任务，天然保证每个任务只被处理一次
• 独立Driver实例：每个线程启动时创建自己的ChromeDriver，处理完任务后关闭，彻底避免多线程共用Driver的冲突问题
• 写入锁机制：多个线程向同一文件写结果时，用锁控制访问，防止数据错乱

完整实现代码 #

import openpyxl
import time
from seleniumwire import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from selenium.common.exceptions import WebDriverException
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from threading import Thread, Lock
import queue

# 全局配置
THREAD_COUNT = 5  # 设定5个线程（对应5个ChromeDriver）
RESULT_FILE = "result.csv"
LOCK = Lock()
TASK_QUEUE = queue.Queue()

def init_driver():
    # 初始化ChromeDriver，可根据需求添加配置
    chrome_options = Options()
    # 可选：无头模式，减少资源占用
    # chrome_options.add_argument("--headless=new")
    chrome_options.add_argument("--disable-gpu")
    chrome_options.add_argument("--no-sandbox")
    driver = webdriver.Chrome(options=chrome_options)
    driver.header_overrides = {
        'USER-AGENT':'Mozilla/5.0 (Linux; Android 10; SM-G981B) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.162 Mobile Safari/537.36',
    }
    return driver

def worker():
    # 线程工作函数：从队列取任务，处理，写入结果
    driver = init_driver()
    url = "https://testautomationpractice.blogspot.com/"
    while not TASK_QUEUE.empty():
        try:
            num = TASK_QUEUE.get(timeout=1)
            print(f"线程{Thread.current_thread().name}处理数据：{num}")
            
            driver.get(url)
            # 等待搜索框并输入数据
            WebDriverWait(driver, 60).until(
                EC.element_to_be_clickable((By.CLASS_NAME, 'wikipedia-search-input'))
            ).send_keys(num)
            time.sleep(3)
            
            # 这里添加抓取结果的逻辑，示例中假设抓取搜索结果文本
            # 比如：result = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, 'wikipedia-search-result'))).text
            result = f"{num},处理完成"  # 示例结果
            
            # 线程安全写入结果文件
            with LOCK:
                with open(RESULT_FILE, "a", encoding="utf-8") as f:
                    f.write(f"{result}\n")
            
            TASK_QUEUE.task_done()
        except queue.Empty:
            break
        except WebDriverException as e:
            print(f"线程{Thread.current_thread().name}出错：{str(e)}")
            TASK_QUEUE.task_done()
    # 关闭当前线程的Driver
    driver.quit()
    print(f"线程{Thread.current_thread().name}结束")

def load_tasks():
    # 从Excel加载任务到队列
    wb = openpyxl.load_workbook("test.xlsx")
    sheet = wb.active
    for row in sheet.iter_rows(min_row=1, min_col=1, max_col=1):
        for cell in row:
            if cell.value is not None:  # 跳过空单元格
                TASK_QUEUE.put(cell.value)

if __name__ == "__main__":
    # 初始化任务队列
    load_tasks()
    # 创建并启动线程
    threads = []
    for i in range(THREAD_COUNT):
        t = Thread(name=f"Thread-{i+1}", target=worker)
        t.start()
        threads.append(t)
    # 等待所有任务完成
    TASK_QUEUE.join()
    # 等待所有线程结束
    for t in threads:
        t.join()
    print("所有任务处理完成")

注意事项 #

1. 确保ChromeDriver版本与本地Chrome浏览器版本一致，避免启动失败
2. 线程数可根据机器性能调整，不是越多越好，5个是合理的起步值
3. 如果不需要可视化浏览器，可开启无头模式（代码中已注释），节省系统资源
4. 抓取逻辑部分需要根据实际需求修改，示例中仅模拟了输入操作，需补充结果提取代码
5. 若数据量极大，也可以改为每个线程写入独立CSV文件，最后用脚本合并，但单文件加锁的方式更简洁高效

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