本文详解如何在 seleniumbase 中通过多线程为每个 webdriver 实例分配唯一代理（支持带认证的 http/socks 代理），避免代理复用导致 ip 冲突，并确保驱动隔离、资源安全销毁。
本文详解如何在 seleniumbase 中通过多线程为每个 webdriver 实例分配唯一代理（支持带认证的 http/socks 代理），避免代理复用导致 ip 冲突，并确保驱动隔离、资源安全销毁。
在 SeleniumBase 中实现「每线程一驱动一代理」的关键障碍，是其默认代理机制会复用同一个 ZIP 扩展文件（用于注入代理认证凭据），导致多线程下所有 Chrome 实例实际共用首个代理配置。解决方案的核心在于显式启用 multi_proxy=True 参数——它将为每个 get_new_driver() 调用生成唯一命名的代理扩展 ZIP 文件，彻底隔离代理上下文。
以下是一个不依赖 pytest、纯 Python concurrent.futures 的生产级示例，完全适配您提出的两阶段任务流（先并发抓取列表页链接，再销毁旧驱动、按新代理池并发解析商品页）：
import concurrent.futures
import time
from seleniumbase import Driver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
# ✅ 阶段1：列表页代理池（每页独立代理）
pool_proxies_for_pages = [
"user1:pass1@proxy1.example.com:8080",
"user2:pass2@proxy2.example.com:8080",
"user3:pass3@proxy3.example.com:8080"
]
# ✅ 阶段2：商品页代理池（与列表页解耦，同样独立）
pool_proxies_for_products = [
"user4:pass4@proxy4.example.com:8080",
"user5:pass5@proxy5.example.com:8080",
"user6:pass6@proxy6.example.com:8080"
]
def parse_list_page(page_index: int, proxy: str) -> list:
"""并发抓取单个列表页的所有商品链接，返回 href 列表"""
driver = Driver(
uc=True,
proxy=proxy,
multi_proxy=True, # ? 强制启用多代理隔离！必加！
agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
headless=True, # 生产环境建议启用
incognito=True
)
try:
url = f"https://www.ebay.com/e/_electronics/shop-all-ebay-refurbished-cell-phones?_pgn={page_index + 1}"
driver.get(url)
# 智能等待 + 滚动触发懒加载
wait = WebDriverWait(driver, 15)
wait.until(EC.presence_of_element_located((By.TAG_NAME, "body")))
driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
time.sleep(3)
# 提取所有商品链接（更健壮的 XPath）
links = []
elements = driver.find_elements(By.XPATH, "//a[contains(@href, '/itm/') and @tabindex='-1']")
for elem in elements[:20]: # 限制数量防超时
href = elem.get_attribute("href")
if href and "ebay.com/itm/" in href:
links.append(href)
return links
finally:
driver.quit() # ✅ 确保驱动及时释放
def parse_product_page(link: str, proxy: str, page_idx: int):
"""使用指定代理解析单个商品页（可扩展业务逻辑）"""
driver = Driver(
uc=True,
proxy=proxy,
multi_proxy=True, # ? 同样必须开启
headless=True,
incognito=True
)
try:
driver.get(link)
# 示例：提取标题
title = driver.find_element(By.TAG_NAME, "h1").text[:100] if driver.find_elements(By.TAG_NAME, "h1") else "N/A"
print(f"[Page-{page_idx}] Parsed {link[:50]}... → '{title}'")
except Exception as e:
print(f"[Error on {link}]: {e}")
finally:
driver.quit()
# ? 主流程：两阶段并发执行
if __name__ == "__main__":
# 阶段1：并发获取所有列表页链接
print("? Stage 1: Fetching product links from list pages...")
with concurrent.futures.ThreadPoolExecutor(max_workers=len(pool_proxies_for_pages)) as executor:
list_futures = [
executor.submit(parse_list_page, i, pool_proxies_for_pages[i])
for i in range(len(pool_proxies_for_pages))
]
all_links = []
for future in concurrent.futures.as_completed(list_futures):
links = future.result()
all_links.extend(links)
print(f"✅ Collected {len(all_links)} product links")
# 阶段2：销毁旧驱动后，按新代理池并发解析商品页
print("? Stage 2: Parsing product pages with dedicated proxies...")
with concurrent.futures.ThreadPoolExecutor(max_workers=len(pool_proxies_for_products)) as executor:
# 将链接轮询分配给代理（避免某代理过载）
for idx, link in enumerate(all_links):
proxy = pool_proxies_for_products[idx % len(pool_proxies_for_products)]
executor.submit(parse_product_page, link, proxy, idx)
⚠️ 关键注意事项：
- multi_proxy=True 是硬性要求：缺失该参数将导致所有线程共享同一代理 ZIP，IP 地址必然重复；
- 代理格式必须规范：user:pass@host:port（HTTP）或 user:pass@host:port（SOCKS5 需额外指定 --proxy-server=socks5://host:port，SeleniumBase v4.19+ 原生支持）；
- 资源清理不可省略：每个 Driver() 实例必须调用 .quit()，否则残留进程会耗尽系统句柄；
- 避免全局共享状态：切勿在函数外定义 driver 或复用 WebDriverWait 实例，所有对象应在函数内创建；
- 线程安全增强：若遇到 Chrome failed to start 等竞争错误，可在 Driver() 中添加 disable_gpu=True, no_sandbox=True（仅限 Linux 服务器）。
通过以上结构，您即可实现真正意义上的「每线程专属代理 + 驱动生命周期自治」，完美支撑高并发、高匿名性的网页采集任务。
