本文详解如何在 seleniumbase 中通过多线程为每个 webdriver 实例分配唯一代理（含认证），避免代理复用导致 ip 冲突，并提供可直接运行的生产级示例代码。
本文详解如何在 seleniumbase 中通过多线程为每个 webdriver 实例分配唯一代理（含认证），避免代理复用导致 ip 冲突，并提供可直接运行的生产级示例代码。
在 SeleniumBase 中实现「每线程独占代理」的核心难点在于：默认情况下，proxy= 参数会触发内部代理认证 ZIP 扩展的全局复用机制——即所有线程共用同一个 proxy_auth_extension.zip 文件，导致所有浏览器最终使用相同代理（甚至相同认证凭据），彻底失去并发隔离性。
✅ 正确解法是显式启用 multi_proxy=True 参数。该参数会为每个 get_new_driver() 调用生成唯一命名的代理扩展 ZIP 文件（如 proxy_auth_extension_1a2b3c.zip），确保各线程驱动完全独立，互不干扰。
以下是一个基于 concurrent.futures.ThreadPoolExecutor 的完整、可运行的教程示例，严格对应你的需求场景（分两阶段：先并发爬取列表页获取链接，再销毁旧驱动、新建带独立代理的驱动并发解析商品页）：
import concurrent.futures
import time
from seleniumbase import Driver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.action_chains import ActionChains
# ✅ 代理池：列表页阶段 & 商品页阶段各自独立
pool_proxies_for_pages = [
"user1:pass1@proxy1.example.com:8080",
"user2:pass2@proxy2.example.com:8080",
"user3:pass3@proxy3.example.com:8080"
]
pool_proxies_for_products = [
"user4:pass4@proxy4.example.com:8080",
"user5:pass5@proxy5.example.com:8080",
"user6:pass6@proxy6.example.com:8080"
]
def create_driver_with_unique_proxy(proxy_url):
"""关键：必须传入 multi_proxy=True"""
return Driver(
uc=True,
proxy=proxy_url,
multi_proxy=True, # ? 强制启用多代理隔离模式
agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
headless=False, # 开发时建议设为 False，便于调试
incognito=True
)
def parse_page(page_index, proxy):
"""阶段一：并发抓取列表页链接"""
driver = create_driver_with_unique_proxy(proxy)
try:
url = f"https://www.ebay.com/e/_electronics/shop-all-ebay-refurbished-cell-phones?_pgn={page_index + 1}"
driver.get(url)
# 模拟滚动到底部（增强反检测）
actions = ActionChains(driver)
actions.send_keys(Keys.END).perform()
time.sleep(5)
# 提取商品链接（XPath 示例，按需调整）
links = []
elements = driver.find_elements(By.XPATH, "//a[@tabindex='-1']")
for elem in elements[:10]: # 限制数量便于测试
href = elem.get_attribute("href")
if href and "item" in href:
links.append(href)
print(f"[列表页 {page_index}] 获取到 {len(links)} 个商品链接")
return links
finally:
driver.quit() # ✅ 立即释放资源
def parse_product_batch(product_links, proxy_index):
"""阶段二：为一批商品链接创建新驱动并发解析"""
proxy = pool_proxies_for_products[proxy_index % len(pool_proxies_for_products)]
driver = create_driver_with_unique_proxy(proxy)
try:
for link in product_links:
driver.get(link)
# ✅ 在此处添加你的商品信息提取逻辑
# 例如：title = driver.find_element(By.ID, "itemTitle").text
time.sleep(1) # 避免请求过快被限流
print(f"[商品解析] 线程 {proxy_index} 完成 {len(product_links)} 个页面")
finally:
driver.quit()
# ? 主执行流程
if __name__ == "__main__":
# 第一阶段：并发获取列表页链接
all_links = []
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
futures = [
executor.submit(parse_page, i, pool_proxies_for_pages[i])
for i in range(min(3, len(pool_proxies_for_pages)))
]
for future in concurrent.futures.as_completed(futures):
links = future.result()
if links:
all_links.extend(links)
print(f"✅ 第一阶段完成，共收集 {len(all_links)} 个商品链接")
# 第二阶段：将链接分组，每组分配给一个带独立代理的新驱动
batch_size = max(1, len(all_links) // len(pool_proxies_for_products))
batches = [
all_links[i:i + batch_size]
for i in range(0, len(all_links), batch_size)
]
print(f"✅ 准备启动 {len(batches)} 个并发商品解析任务")
with concurrent.futures.ThreadPoolExecutor(max_workers=len(pool_proxies_for_products)) as executor:
futures = [
executor.submit(parse_product_batch, batch, idx)
for idx, batch in enumerate(batches)
]
concurrent.futures.wait(futures)
print("? 全部任务执行完毕！")
⚠️ 关键注意事项：
- multi_proxy=True 是强制必需项，缺省值 False 会导致所有线程共享同一代理扩展 ZIP，IP 必然重复；
- 每次调用 Driver(...) 或 get_new_driver(...) 均需传入 multi_proxy=True，不可只在初始化时设置；
- 若使用 pytest-xdist，请改用 --seleniumbase-multi-proxy 命令行参数（详见 SeleniumBase Issue #2478），避免线程间临时文件竞争；
- 生产环境建议添加异常重试、超时控制（timeout=30）、User-Agent 轮换及请求间隔，以提升稳定性；
- 代理格式必须为标准 user:pass@host:port，否则 multi_proxy=True 将静默失效。
通过以上结构化实现，你可真正达成「每个线程驱动绑定唯一代理、全生命周期隔离、两阶段驱动按需重建」的目标，兼顾性能、稳定与可维护性。
