Python+Selenium多线程网页爬取报错问题求助

核心问题 #

你遇到的两种多线程报错，本质都是WebDriver实例不支持多线程共享。单线程时一个driver对应一个独立会话，操作串行无冲突；但多线程共用同一个driver时，线程间的页面跳转、刷新操作会互相干扰：

• ThreadPoolExecutor后续任务执行时，driver已被前一个任务的页面操作改变状态，导致找不到tbody元素（返回None）
• threading.Thread的"stale element"错误，是因为前一个线程操作后页面重新加载，后一个线程持有的元素引用已失效

解决方法 #

每个线程必须创建独立的WebDriver实例，彻底避免线程间的会话干扰。以下是两种修复后的实现示例：

1. ThreadPoolExecutor 修复版 #

from concurrent.futures import ThreadPoolExecutor
from selenium import webdriver
from selenium.webdriver.common.by import By

def crawl_license(license_id):
    # 每个任务独立初始化driver
    driver = webdriver.Chrome()
    try:
        driver.get("指定网站URL")
        # 执行查询操作：输入编号、点击查询
        input_box = driver.find_element(By.ID, "查询输入框ID")
        input_box.clear()
        input_box.send_keys(license_id)
        driver.find_element(By.ID, "查询按钮ID").click()
        # 提取数据
        tbody = driver.find_element(By.TAG_NAME, "tbody")
        data = tbody.text
        return (license_id, data)
    except Exception as e:
        print(f"编号{license_id}爬取失败: {str(e)}")
        return (license_id, None)
    finally:
        driver.quit() # 每个任务结束后关闭driver

# 执行多线程爬取
license_ids = ["编号1", "编号2", ...] # 你的3万个编号列表
with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(crawl_license, license_ids))

2. threading.Thread 修复版 #

import threading
from selenium import webdriver
from selenium.webdriver.common.by import By

def crawl_license(license_id, result_list):
    driver = webdriver.Chrome()
    try:
        driver.get("指定网站URL")
        input_box = driver.find_element(By.ID, "查询输入框ID")
        input_box.clear()
        input_box.send_keys(license_id)
        driver.find_element(By.ID, "查询按钮ID").click()
        tbody = driver.find_element(By.TAG_NAME, "tbody")
        data = tbody.text
        result_list.append((license_id, data))
    except Exception as e:
        print(f"编号{license_id}爬取失败: {str(e)}")
        result_list.append((license_id, None))
    finally:
        driver.quit()

# 执行多线程爬取
license_ids = ["编号1", "编号2", ...]
results = []
threads = []
for lic_id in license_ids:
    t = threading.Thread(target=crawl_license, args=(lic_id, results))
    threads.append(t)
    t.start()

# 等待所有线程完成
for t in threads:
    t.join()

额外优化建议 #

• 控制线程数量：不要设置过大的max_workers，避免触发网站反爬机制，建议5-10个线程即可
• 添加显式等待：替换直接find_element的方式，用WebDriverWait等待元素加载完成，避免页面加载慢导致的元素找不到问题：

  from selenium.webdriver.support.ui import WebDriverWait
  from selenium.webdriver.support import expected_conditions as EC
  
  # 等待输入框出现，最多等10秒
  input_box = WebDriverWait(driver, 10).until(
      EC.presence_of_element_located((By.ID, "查询输入框ID"))
  )
  

• 细化异常处理：针对元素找不到、请求超时等不同错误做针对性处理，方便后续排查问题

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