SeleniumBase启用uc=True时多线程失效问题求助

核心原因 #

UC模式（Undetected Chromedriver）为绕过反爬检测，会复用部分Chrome全局配置状态，而Python多线程共享同一内存空间，导致多个UC实例的会话、配置互相干扰，引发页面加载异常。

可行解决方案 #

1. 改用多进程+共享状态管理 #

多进程拥有独立内存空间，能彻底隔离每个UC实例的运行环境，同时用multiprocessing.Manager实现代理列表的跨进程共享：

• 替换ThreadPoolExecutor为ProcessPoolExecutor
• 用Manager().list()创建可跨进程修改的available_proxies列表
• 为每个进程的SB实例分配独立的用户数据目录，避免会话冲突

2. 强制隔离Chrome启动参数 #

在初始化SB时，添加专属Chrome选项，确保每个实例完全独立：

• 设置user_data_dir为唯一路径（比如结合进程ID生成）
• 添加--no-sandbox、--disable-dev-shm-usage等参数避免权限/资源冲突

3. 移除递归调用逻辑 #

原代码中递归重试的逻辑在多进程环境下会导致进程资源泄漏，改为重新提交任务到进程池的方式实现重试。

修改后的完整代码 #

from seleniumbase import SB
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import Manager
import os
import tempfile
import shutil

def process_request(proxy, email, available_proxies, executor):
    # 为每个进程生成唯一的用户数据目录
    temp_dir = tempfile.mkdtemp(prefix=f"sb_uc_{os.getpid()}_")
    chrome_options = [
        "--no-sandbox",
        "--disable-dev-shm-usage",
        f"--user-data-dir={temp_dir}"
    ]
    
    try:
        with SB(uc=True, headless=False, proxy=proxy, chrome_options=chrome_options) as sb:
            sb.open("https://google.com/")

            try:
                sb.assert_element('div[class="css-y8g9ca"]', timeout=4)
            except Exception as e:
                print(e)
                print(f"Bad proxy: {proxy}")
                # 线程安全地移除坏代理
                if proxy in available_proxies:
                    available_proxies.remove(proxy)
                
                if available_proxies:
                    new_proxy = available_proxies[0]
                    print(f"Retrying with a new proxy: {new_proxy}")
                    # 重新提交任务到进程池，代替递归
                    executor.submit(process_request, new_proxy, email, available_proxies, executor)
                else:
                    print("No more proxies available.")
                return

            sb.type('input[id="username"]', email)
            sb.click('button:contains("Next")')

            try:
                sb.assert_element('div[class="css-i76c7y"]', timeout=5)
                print(f"Valid email {email} with proxy {proxy}")
            except Exception as e:
                print(e)
                print(f"Invalid email {email}")
    finally:
        # 清理临时用户目录
        if os.path.exists(temp_dir):
            shutil.rmtree(temp_dir, ignore_errors=True)

def main():
    proxies = [i.rstrip() for i in open("proxies.txt", "r+")]
    emails = [i.rstrip() for i in open("emails.txt", "r+")]

    # 用Manager创建跨进程共享的代理列表
    with Manager() as manager:
        available_proxies = manager.list(proxies)
        num_processes = 3

        with ProcessPoolExecutor(max_workers=num_processes) as executor:
            futures = []
            # 提交初始任务
            for proxy, email in zip(proxies, emails):
                futures.append(executor.submit(process_request, proxy, email, available_proxies, executor))
            
            # 等待所有任务完成
            for future in futures:
                try:
                    future.result()
                except Exception as e:
                    print(f"Task failed: {e}")

if __name__ == "__main__":
    main()

关键修改说明 #

• 用ProcessPoolExecutor替代线程池，彻底隔离每个UC实例的运行环境
• 为每个进程生成临时user_data_dir，避免Chrome会话共享
• 使用multiprocessing.Manager实现代理列表的跨进程安全修改
• 移除递归调用，改用提交新任务到进程池的方式实现重试
• 添加临时目录清理逻辑，避免磁盘资源占用

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