Python迭代中创建线程实现网站QA测试并行执行需求

首先明确核心前提：WebDriver实例不是线程安全的，绝对不能在多个线程里共用同一个driver，每个线程必须单独初始化自己的Chrome驱动，否则会出现各种不可预期的异常。下面给出两种实用的实现方案：

方案一：用concurrent.futures.ThreadPoolExecutor（推荐，简洁易管理） #

这个模块可以自动帮你管理线程池，还能方便追踪任务完成状态：

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

# 定义每个线程要执行的测试任务
def run_test(iteration_num):
    # 每个线程单独初始化Chrome驱动
    options = webdriver.ChromeOptions()
    # 可按需添加配置，比如无头模式：options.add_argument("--headless=new")
    driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=options)
    
    try:
        driver.get("你的网站URL")
        print(f"正在执行测试: {iteration_num}")
        # 这里可以添加你的QA测试逻辑，比如点击元素、表单提交、截图等
        time.sleep(2)  # 模拟测试操作耗时
        print(f"测试 {iteration_num} 执行完成")
    except Exception as e:
        print(f"测试 {iteration_num} 执行失败: {str(e)}")
    finally:
        # 无论成功失败，必须关闭driver释放资源
        driver.quit()

if __name__ == "__main__":
    total_tests = 10
    # 线程池大小，根据设备性能调整（比如CPU核心数*2）
    max_workers = 10

    # 创建线程池并批量提交任务
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        executor.map(run_test, range(1, total_tests + 1))

方案二：用threading模块手动管理线程 #

如果你更习惯手动控制线程生命周期，可以用这种方式：

import threading
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
import time

def run_test(iteration_num):
    options = webdriver.ChromeOptions()
    driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=options)
    try:
        driver.get("你的网站URL")
        print(f"正在执行测试: {iteration_num}")
        time.sleep(2)
        print(f"测试 {iteration_num} 执行完成")
    except Exception as e:
        print(f"测试 {iteration_num} 执行失败: {str(e)}")
    finally:
        driver.quit()

if __name__ == "__main__":
    total_tests = 10
    threads = []

    # 创建并启动所有线程
    for i in range(1, total_tests + 1):
        thread = threading.Thread(target=run_test, args=(i,))
        threads.append(thread)
        thread.start()

    # 等待所有线程执行完毕
    for thread in threads:
        thread.join()

    print("所有测试任务已完成")

关键注意事项 #

• 并发数调整：根据CPU、内存和网络情况调整线程池大小，不要盲目设得过大，避免系统资源耗尽或给网站服务器造成过大压力。
• 资源清理：必须在finally块里调用driver.quit()，否则会残留Chrome进程占用内存。
• 测试扩展：可以在run_test函数里添加更多QA逻辑，比如元素定位、断言、生成测试报告/样本文件等。
• 异常处理：添加异常捕获可以避免单个线程的错误导致整个程序崩溃，同时方便排查问题。

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