2026年Selenium爬虫代理IP集成实战:从环境搭建到项目落地

简介: 本文直击 Selenium 代理集成痛点:先破“`--proxy-server` 万能论”误区,再系统梳理2026年三大实战方案——无认证直连、账密代理插件(Manifest V3兼容)、隧道代理,并附SOCKS5接法、IP轮换、反检测、并发架构及可运行模板。专为舆情监测、价格爬取等合规动态渲染场景而写,踩坑经验全来自真实项目。

先说结论,再说为什么。

Selenium 接代理这事儿,2026 年了还有一堆人在踩 2019 年的坑:Chrome Options 里塞个 --proxy-server 就觉得搞定了,跑两小时发现带认证的代理根本没生效,IP 还是自己的出口。今年 4 月我们工作室接了一单舆情监测的活儿,要用 Selenium 渲染动态页面再提数据,代理这层我重新梳理了一遍,把几种集成方式、踩过的坑、以及一套能直接跑的项目模板都整理出来。

这篇写给要用 Selenium 做合规数据采集的工程师:公开网页舆情、价格监测、学术信息聚合这类场景。

环境这一层,2026 年有几个变化

往下拆一层看,先把开发环境理清楚。今年的 Selenium 生态跟两三年前比,有几个值得注意的变化。

Python + Selenium 4.x + WebDriver Manager

# Python 3.10+ 推荐,我们工作室统一用 3.11
pip install selenium==4.25.0
pip install webdriver-manager==4.0.2

Selenium 4 之后 WebDriver 的管理方式改了,不用再手动下 chromedriver 丢到 PATH 里了。webdriver-manager 会自动匹配你本地 Chrome 的版本,省了一堆版本不对的报错。

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

service = Service(ChromeDriverManager().install())
driver = webdriver.Chrome(service=service)
driver.get("https://httpbin.org/ip")
print(driver.page_source)
driver.quit()

跑通这段,说明你的基础环境没问题。

还有一点:Chrome 从 115 版本开始把 chromedriver 的发布渠道从 ChromeDriver 官网迁到了 Chrome for Testing,webdriver-manager 新版已经适配了这个变更,但如果你用的是老版本的 manager,大概率会报 404——升到 4.0 以上就行。

代理集成:三种方式,各有各的适用边界

这地方很多人踩坑——不是所有代理都能用同一种方式接。我按实际使用频率排,从最简单的开始。

方式一:Chrome 启动参数直接注入(无认证代理)

最简单的一种,适合不需要账密认证的代理——比如你用的是白名单鉴权,已经在代理服务商后台把你的出口 IP 加了白名单。

from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from webdriver_manager.chrome import ChromeDriverManager

def create_driver_with_proxy(proxy_address):
    """
    proxy_address 格式: "host:port"
    例如: "222.xxx.xxx.xx:30001"
    """
    chrome_options = Options()
    chrome_options.add_argument(f"--proxy-server=http://{proxy_address}")

    # 2026 年的 Chrome 默认开了一些安全策略,跑采集建议关掉这几个
    chrome_options.add_argument("--disable-blink-features=AutomationControlled")
    chrome_options.add_argument("--no-first-run")
    chrome_options.add_argument("--no-default-browser-check")

    service = Service(ChromeDriverManager().install())
    driver = webdriver.Chrome(service=service, options=chrome_options)
    return driver

# 用法
driver = create_driver_with_proxy("你的代理IP:端口")
driver.get("https://httpbin.org/ip")
print(driver.find_element("tag name", "body").text)
driver.quit()

这种方式的边界很明确:只能走 HTTP/HTTPS 代理,不支持 SOCKS5,不支持账密认证。白名单鉴权的场景用它就够了,简单直接。

方式二:Chrome 插件注入认证信息(带账密的代理)

这是实际项目里用得最多的方式。大多数代理服务商提供的都是账密认证——你拿到一个 host:port:username:password,需要在请求时带上用户名和密码。

问题恰恰出在这儿。Selenium 的 Chrome WebDriver 原生不支持在启动参数里传代理的用户名密码,--proxy-server 那个参数只接 host:port,没有账密的位置。2019 年有人用 driver.switch_to.alert 去接 HTTP 407 弹窗输入密码,那个方案在 Chrome 100 以后就废了。

现在的主流方案是动态生成一个 Chrome 插件,用插件的 webRequest API 拦截 407 响应并自动填入账密。代码不长,但细节多,我直接给能跑的版本:

import os
import zipfile
import tempfile
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from webdriver_manager.chrome import ChromeDriverManager

def create_proxy_auth_extension(proxy_host, proxy_port, proxy_user, proxy_pass):
    """动态生成一个 Chrome 插件来处理代理认证"""

    manifest_json = """{
        "version": "1.0.0",
        "manifest_version": 3,
        "name": "Proxy Auth Helper",
        "permissions": ["webRequest", "webRequestAuthProvider"],
        "host_permissions": ["<all_urls>"],
        "background": {
            "service_worker": "background.js"
        }
    }"""

    background_js = """
    chrome.webRequest.onAuthRequired.addListener(
        function(details, callbackFn) {
            callbackFn({
                authCredentials: {
                    username: "%s",
                    password: "%s"
                }
            });
        },
        {urls: ["<all_urls>"]},
        ["asyncBlocking"]
    );
    """ % (proxy_user, proxy_pass)

    # 打包成 .crx 临时文件
    plugin_dir = tempfile.mkdtemp()
    plugin_file = os.path.join(plugin_dir, "proxy_auth.zip")

    with zipfile.ZipFile(plugin_file, "w") as zp:
        zp.writestr("manifest.json", manifest_json)
        zp.writestr("background.js", background_js)

    return plugin_file

def create_driver_with_auth_proxy(host, port, username, password):
    """创建带账密认证代理的 WebDriver"""

    chrome_options = Options()

    # 加载认证插件
    plugin_path = create_proxy_auth_extension(host, port, username, password)
    chrome_options.add_extension(plugin_path)

    # 设置代理地址
    chrome_options.add_argument(f"--proxy-server=http://{host}:{port}")
    chrome_options.add_argument("--disable-blink-features=AutomationControlled")

    service = Service(ChromeDriverManager().install())
    driver = webdriver.Chrome(service=service, options=chrome_options)
    return driver

# 用法
driver = create_driver_with_auth_proxy(
    host="代理地址",
    port="端口",
    username="你的账号",
    password="你的密码"
)
driver.get("https://httpbin.org/ip")
print(driver.find_element("tag name", "body").text)
driver.quit()

这里有个坑要说一下。2025 年中开始 Chrome 全面推 Manifest V3,老版本的插件用的是 chrome.webRequest.onAuthRequiredblocking 模式,V3 里要改成 asyncBlocking 加 callback 写法,不然插件加载直接报错。上面的代码已经是 V3 兼容的写法。

我在今年 3 月初测试的时候,还遇到一个问题:Chrome 129 开始对 service_worker 的生命周期管理更严了,如果代理认证响应太慢(超过 30 秒),service worker 可能会被 Chrome 杀掉,导致后续请求认证失败。解决办法是在 background.js 里加一个 keep-alive 心跳,但对大多数正常延迟的代理服务来说,这个问题不太会触发。

方式三:隧道代理——最省事的方案

隧道代理的核心思路是:代理服务商那边帮你做 IP 轮换,你这边只管往一个固定的网关地址发请求。每次请求过去,出口 IP 自动换。

对 Selenium 来说,隧道代理的集成方式和方式一、方式二没有本质区别——它本质上还是一个 host:port(可能带账密)。区别在于你不需要自己写 IP 轮换逻辑了。

# 隧道代理集成(以账密认证为例)
# 你拿到的隧道网关通常是一个固定地址,比如 tunnel.example.com:12345
# 每次请求经过这个网关,出口 IP 自动切换

driver = create_driver_with_auth_proxy(
    host="tunnel.example.com",
    port="12345",
    username="your_tunnel_user",
    password="your_tunnel_pass"
)

# 验证 IP 是否在变化
import time
for i in range(5):
    driver.get("https://httpbin.org/ip")
    body = driver.find_element("tag name", "body").text
    print(f"第 {i+1} 次请求,出口 IP:{body}")
    time.sleep(2)

driver.quit()

我们工作室从去年三季度起在用隧道代理跑中文网页语料,隧道这种模式确实省心,不用自己管 IP 池、不用写轮换调度,把精力放在业务逻辑上就行。

SOCKS5 代理的接法

单独拎出来说,因为 SOCKS5 的接法跟 HTTP 代理不太一样。

chrome_options = Options()

# SOCKS5 代理直接用启动参数
chrome_options.add_argument("--proxy-server=socks5://127.0.0.1:1080")

# 如果需要 DNS 也走代理(防 DNS 泄露)
chrome_options.add_argument("--host-resolver-rules=MAP * ~NOTFOUND, EXCLUDE 127.0.0.1")

service = Service(ChromeDriverManager().install())
driver = webdriver.Chrome(service=service, options=chrome_options)

SOCKS5 的好处是协议层更底层,TCP 层面的流量都走代理。只不过,SOCKS5 带认证的场景在 Selenium 里同样要走插件方式,原理跟 HTTP 代理的认证插件一样,改一下协议头就行。

IP 轮换:不用隧道代理时怎么自己做

如果你用的是短效代理或者按量提取的 IP 池,每次拿到一批 IP,需要自己做轮换。

这块的核心思路是:不要在同一个 WebDriver 实例里换代理。Chrome 启动后代理是绑死的,想换 IP 就得重启 WebDriver。所以轮换逻辑本质上是一个"创建 → 使用 → 销毁 → 重建"的循环。

import requests
import time
import json

class ProxyRotator:
    """代理 IP 轮换管理器"""

    def __init__(self, api_url):
        """
        api_url: 代理服务商提供的 API 提取地址
        大多数厂商都提供 HTTP API,调一次返回一批 IP
        """
        self.api_url = api_url
        self.ip_pool = []
        self.current_index = 0

    def fetch_proxies(self, count=10):
        """从 API 提取一批新 IP"""
        try:
            resp = requests.get(f"{self.api_url}&count={count}", timeout=10)
            data = resp.json()
            # 不同厂商的返回格式不一样,这里做个通用解析
            # 常见格式:{"code": 0, "data": [{"ip": "x.x.x.x", "port": 1234}, ...]}
            if data.get("code") == 0:
                self.ip_pool = data.get("data", [])
                self.current_index = 0
                print(f"提取到 {len(self.ip_pool)} 个 IP")
            else:
                print(f"提取失败:{data.get('msg', '未知错误')}")
        except Exception as e:
            print(f"API 调用异常:{e}")

    def get_next_proxy(self):
        """获取下一个可用 IP"""
        if not self.ip_pool or self.current_index >= len(self.ip_pool):
            self.fetch_proxies()

        if not self.ip_pool:
            return None

        proxy = self.ip_pool[self.current_index]
        self.current_index += 1
        return f"{proxy['ip']}:{proxy['port']}"

def crawl_with_rotation(urls, proxy_api_url):
    """带 IP 轮换的采集主循环"""

    rotator = ProxyRotator(proxy_api_url)
    results = []

    for url in urls:
        proxy_addr = rotator.get_next_proxy()
        if not proxy_addr:
            print("IP 池耗尽,等待 30 秒后重试")
            time.sleep(30)
            proxy_addr = rotator.get_next_proxy()
            if not proxy_addr:
                print(f"跳过:{url}")
                continue

        driver = None
        try:
            driver = create_driver_with_proxy(proxy_addr)
            driver.set_page_load_timeout(30)
            driver.get(url)

            # 等页面渲染完成(Selenium 的核心价值就在这:能拿到 JS 渲染后的内容)
            time.sleep(3)

            page_content = driver.page_source
            results.append({
   
                "url": url,
                "content": page_content,
                "proxy": proxy_addr,
                "status": "success"
            })
            print(f"成功:{url} | 代理:{proxy_addr}")

        except Exception as e:
            print(f"失败:{url} | 代理:{proxy_addr} | 错误:{e}")
            results.append({
   
                "url": url,
                "proxy": proxy_addr,
                "status": "failed",
                "error": str(e)
            })
        finally:
            if driver:
                driver.quit()

        # 请求间隔,别太激进
        time.sleep(2)

    return results

这段代码有个问题我得承认:每个 URL 都重建一次 WebDriver,开销不小——Chrome 启动 + 关闭大概要 2 到 3 秒。如果你的目标是高吞吐量采集,Selenium 本身就不是最优选择,Playwright 或者纯 HTTP 请求库(requests、httpx)在吞吐量上甩 Selenium 几条街。Selenium 的场景是你必须要浏览器渲染——动态加载的 SPA 页面、需要执行 JS 才能拿到数据的场景。

完整项目模板:舆情页面采集

把前面的模块拼起来,给一个实际可用的项目框架。这个模板是我今年 4 月那单舆情监测项目的精简版,脱敏后拿出来。

"""
Selenium + 代理 IP 舆情采集项目模板
场景:采集目标网站的公开页面内容,用于舆情监测
"""

import os
import json
import time
import logging
import zipfile
import tempfile
from datetime import datetime
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from webdriver_manager.chrome import ChromeDriverManager

# ========== 日志配置 ==========
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler("crawl.log", encoding="utf-8"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)


# ========== 代理认证插件生成 ==========
def build_auth_extension(proxy_host, proxy_port, proxy_user, proxy_pass):
    """生成 Manifest V3 兼容的代理认证插件"""

    manifest = json.dumps({
   
        "version": "1.0.0",
        "manifest_version": 3,
        "name": "Proxy Auth",
        "permissions": ["webRequest", "webRequestAuthProvider"],
        "host_permissions": ["<all_urls>"],
        "background": {
   "service_worker": "background.js"}
    })

    background = """
    chrome.webRequest.onAuthRequired.addListener(
        function(details, callbackFn) {
            callbackFn({
                authCredentials: {
                    username: "%s",
                    password: "%s"
                }
            });
        },
        {urls: ["<all_urls>"]},
        ["asyncBlocking"]
    );
    """ % (proxy_user, proxy_pass)

    ext_dir = tempfile.mkdtemp()
    ext_path = os.path.join(ext_dir, "proxy_auth.zip")
    with zipfile.ZipFile(ext_path, "w") as zp:
        zp.writestr("manifest.json", manifest)
        zp.writestr("background.js", background)
    return ext_path


# ========== WebDriver 工厂 ==========
class DriverFactory:
    """统一管理 WebDriver 的创建和配置"""

    def __init__(self, headless=True):
        self.headless = headless
        self._service = Service(ChromeDriverManager().install())

    def create(self, proxy_host, proxy_port, proxy_user=None, proxy_pass=None):
        """
        创建一个配置好代理的 WebDriver 实例
        支持白名单模式(无账密)和账密认证模式
        """
        opts = Options()

        if self.headless:
            opts.add_argument("--headless=new")

        opts.add_argument(f"--proxy-server=http://{proxy_host}:{proxy_port}")
        opts.add_argument("--disable-blink-features=AutomationControlled")
        opts.add_argument("--disable-gpu")
        opts.add_argument("--no-sandbox")
        opts.add_argument("--disable-dev-shm-usage")
        opts.add_argument("--window-size=1920,1080")
        opts.add_argument("--disable-extensions-except=")

        # 设置 UA,别用默认的 HeadlessChrome
        opts.add_argument(
            "user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/129.0.0.0 Safari/537.36"
        )

        # 账密认证走插件
        if proxy_user and proxy_pass:
            ext_path = build_auth_extension(
                proxy_host, proxy_port, proxy_user, proxy_pass
            )
            opts.add_extension(ext_path)
            # 加了插件就不能开 headless——这是 Chrome 的限制
            # 如果必须 headless + 账密认证,要用 Xvfb 做虚拟显示
            if self.headless:
                logger.warning(
                    "Chrome 插件与 --headless 不完全兼容,"
                    "建议使用 Xvfb 虚拟显示替代 headless 模式"
                )

        driver = webdriver.Chrome(service=self._service, options=opts)
        driver.set_page_load_timeout(30)
        driver.set_script_timeout(15)

        return driver


# ========== 采集核心 ==========
class SentimentCrawler:
    """舆情页面采集器"""

    def __init__(self, proxy_config, headless=True):
        """
        proxy_config 示例:
        {
            "mode": "tunnel",  # tunnel / rotate
            "host": "tunnel.example.com",
            "port": "12345",
            "username": "user",
            "password": "pass",
            "api_url": None  # rotate 模式下用
        }
        """
        self.proxy_config = proxy_config
        self.factory = DriverFactory(headless=headless)
        self.results = []

    def _create_driver(self, proxy_addr=None):
        """根据配置创建 driver"""
        cfg = self.proxy_config

        if proxy_addr:
            host, port = proxy_addr.split(":")
            return self.factory.create(host, port)

        return self.factory.create(
            cfg["host"], cfg["port"],
            cfg.get("username"), cfg.get("password")
        )

    def crawl_page(self, url, wait_selector=None, wait_timeout=10):
        """
        采集单个页面
        wait_selector: 等待某个元素出现再提取(处理动态加载)
        """
        driver = None
        try:
            driver = self._create_driver()
            logger.info(f"正在采集:{url}")

            driver.get(url)

            # 如果指定了等待元素,用显式等待
            if wait_selector:
                WebDriverWait(driver, wait_timeout).until(
                    EC.presence_of_element_located((By.CSS_SELECTOR, wait_selector))
                )
            else:
                time.sleep(3)  # 兜底等待

            # 提取页面信息
            result = {
   
                "url": url,
                "title": driver.title,
                "content": driver.page_source,
                "timestamp": datetime.now().isoformat(),
                "status": "success"
            }

            # 尝试提取正文(通用方案,实际项目按目标站点定制)
            try:
                body_text = driver.find_element(By.TAG_NAME, "body").text
                result["text"] = body_text[:5000]  # 截断,防止单条过大
            except Exception:
                result["text"] = ""

            self.results.append(result)
            logger.info(f"采集成功:{url} | 标题:{result['title'][:50]}")
            return result

        except Exception as e:
            logger.error(f"采集失败:{url} | 错误:{e}")
            self.results.append({
   
                "url": url,
                "status": "failed",
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            })
            return None
        finally:
            if driver:
                driver.quit()

    def crawl_batch(self, urls, interval=3, wait_selector=None):
        """批量采集,带间隔控制"""
        logger.info(f"开始批量采集,共 {len(urls)} 个 URL")

        for i, url in enumerate(urls):
            logger.info(f"进度:{i+1}/{len(urls)}")
            self.crawl_page(url, wait_selector=wait_selector)

            if i < len(urls) - 1:
                time.sleep(interval)

        # 汇总
        success = sum(1 for r in self.results if r["status"] == "success")
        logger.info(f"采集完成:成功 {success}/{len(urls)}")

        return self.results

    def save_results(self, filepath="results.json"):
        """保存结果到 JSON"""
        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(self.results, f, ensure_ascii=False, indent=2)
        logger.info(f"结果已保存:{filepath},共 {len(self.results)} 条")


# ========== 使用示例 ==========
if __name__ == "__main__":

    # 配置你的代理(以隧道代理为例)
    proxy_config = {
   
        "mode": "tunnel",
        "host": "你的隧道地址",
        "port": "端口",
        "username": "账号",
        "password": "密码"
    }

    # 目标 URL 列表(公开页面)
    target_urls = [
        "https://httpbin.org/ip",
        "https://httpbin.org/headers",
        "https://httpbin.org/user-agent",
    ]

    # 开跑
    crawler = SentimentCrawler(proxy_config, headless=True)
    results = crawler.crawl_batch(target_urls, interval=5)
    crawler.save_results("sentiment_results.json")

几个实际跑起来会遇到的坑

写到这儿我有点累,喝口水。

下面这些坑不是我从文档里翻的,是今年实际项目里一个一个踩出来的。

1. headless 模式和 Chrome 插件冲突

Chrome 的 --headless=new 模式加载插件有兼容性问题——不是完全不能用,而是有些插件的 service_worker 在 headless 下会静默失败,连日志都不打。如果你的代理是账密认证、必须走插件,建议在服务器上用 Xvfb 做虚拟显示,别用 headless:

# 安装 Xvfb(Ubuntu/Debian)
sudo apt-get install xvfb

# 用 Xvfb 启动你的脚本
xvfb-run --auto-servernum python your_crawler.py

或者用 Python 的 pyvirtualdisplay

from pyvirtualdisplay import Display

display = Display(visible=0, size=(1920, 1080))
display.start()

# 你的 Selenium 代码放这里,不加 --headless
# ...

display.stop()

2. WebDriver 实例泄漏

这个坑是真的痛。如果你的代码在异常路径上没有正确调用 driver.quit(),Chrome 进程不会自己退——跑一晚上第二天早上服务器上挂着 200 个僵尸 Chrome 进程,内存直接拉满。

我现在的做法是用 atexit 模块加一层兜底:

import atexit
import signal

_active_drivers = []

def register_driver(driver):
    _active_drivers.append(driver)

def cleanup_drivers():
    for d in _active_drivers:
        try:
            d.quit()
        except Exception:
            pass
    _active_drivers.clear()

atexit.register(cleanup_drivers)
signal.signal(signal.SIGTERM, lambda *args: (cleanup_drivers(), exit(0)))

加上 try/finally 的双保险,基本不会再泄漏了。

3. 代理超时和页面加载超时的区分

driver.set_page_load_timeout(30) 设的是整个页面的加载超时,不是代理连接超时。如果你的代理本身就连不上,这 30 秒会全部耗完才报错。

想快速判断代理是否可用,可以在创建 WebDriver 之前,先用 requests 库做一次轻量级验活:

import requests

def check_proxy_alive(proxy_addr, timeout=5):
    """快速验证代理是否可达"""
    try:
        resp = requests.get(
            "https://httpbin.org/ip",
            proxies={
   "http": f"http://{proxy_addr}", "https": f"http://{proxy_addr}"},
            timeout=timeout
        )
        return resp.status_code == 200
    except Exception:
        return False

先验活再启动 Chrome,能省不少时间。

4. 目标站点检测到 Selenium 的特征

这个话题能展开讲一整篇,这里只说最基本的几条。Chrome 在 Selenium 控制下会暴露一些特征——navigator.webdriver 属性为 truewindow.chrome.runtime 缺失等。

# 基础的反检测配置
chrome_options.add_argument("--disable-blink-features=AutomationControlled")
chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
chrome_options.add_experimental_option("useAutomationExtension", False)

# 启动后注入 JS 覆盖 webdriver 属性
driver.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument", {
   
    "source": """
        Object.defineProperty(navigator, 'webdriver', {
            get: () => undefined
        });
    """
})

只不过,这里有个前提:这些手段能应对基础检测,但遇到专业的反爬体系(比如某些站点用的行为分析 + 指纹检测),光靠这几行代码远远不够。更核心的是你的请求行为要像真人——请求间隔随机化、滚动页面、鼠标轨迹模拟这些,比改浏览器特征更有效。

并发采集的架构选择

如果单线程不够用,要上并发,Selenium 的并发方案有两条路:

路线一:多进程(推荐)

Selenium 的 WebDriver 不是线程安全的,同一个 driver 实例不能跨线程操作。所以多线程方案基本不用考虑,直接上多进程:

from multiprocessing import Pool

def crawl_single(args):
    """单个采集任务(在子进程里执行)"""
    url, proxy_config = args
    crawler = SentimentCrawler(proxy_config, headless=True)
    return crawler.crawl_page(url)

def crawl_parallel(urls, proxy_config, workers=4):
    """多进程并发采集"""
    args_list = [(url, proxy_config) for url in urls]

    with Pool(processes=workers) as pool:
        results = pool.map(crawl_single, args_list)

    return [r for r in results if r is not None]

4 个 worker 大概要 4 到 6 GB 内存(每个 Chrome 实例 1 到 1.5 GB),服务器配置得跟上。

路线二:Selenium Grid

如果你的采集量级到了每天几万页以上,单机多进程也扛不住了,可以上 Selenium Grid。Grid 是 Selenium 官方的分布式方案,Hub 节点负责调度,Node 节点跑实际的浏览器。

# 2026 年用 Docker 拉 Grid 最方便
docker network create grid

# Hub
docker run -d -p 4442-4444:4442-4444 \
  --net grid --name selenium-hub \
  selenium/hub:4.25.0

# Chrome Node(可以起多个)
docker run -d --net grid \
  -e SE_EVENT_BUS_HOST=selenium-hub \
  -e SE_EVENT_BUS_PUBLISH_PORT=4442 \
  -e SE_EVENT_BUS_SUBSCRIBE_PORT=4443 \
  -e SE_NODE_MAX_SESSIONS=4 \
  selenium/node-chrome:4.25.0

代码端连 Grid:

from selenium import webdriver

def create_remote_driver(grid_url, proxy_addr):
    """连接 Selenium Grid 远程节点"""
    opts = Options()
    opts.add_argument(f"--proxy-server=http://{proxy_addr}")
    opts.add_argument("--disable-blink-features=AutomationControlled")

    driver = webdriver.Remote(
        command_executor=grid_url,
        options=opts
    )
    return driver

# grid_url 通常是 http://hub-ip:4444
driver = create_remote_driver("http://localhost:4444", "代理地址:端口")

只不过,Grid 的方案复杂度不低,运维成本也高。大多数中小规模的采集任务,单机多进程 + 隧道代理完全够用,别过度架构。

这些年的经验浓缩成几句话

回到刚才那个数,我做了十年数据采集,2022 年在字节那会儿,每天过手的页面量级是千万级的。那时候用的不是 Selenium,那个量级 Selenium 根本扛不住,是自研的 Headless Chrome 集群 + 消息队列。

Selenium 的定位是什么?是中小规模、需要浏览器渲染、需要跑 JavaScript 的合规数据采集。日均几千到几万页面,Selenium 很合适;日均十万以上,就得考虑 Playwright(性能更好、原生支持代理认证、多浏览器支持)或者更底层的方案了。

代理 IP 这层,不管你用什么采集框架,核心就三件事:认证方式搞对、IP 轮换做好、异常处理兜住。框架在换,这三件事十年了没变过。

相关文章
|
21天前
|
Linux 程序员 数据格式
【2026最新】Notepad++下载、安装和使用一篇搞定(附中文版安装包)
Notepad++ 是一款免费开源、轻量高效的 Windows 文本编辑器,支持 C/Python/HTML 等 80+ 语言语法高亮、代码折叠、正则替换、编码转换及插件扩展,专为程序员与文本处理用户打造,完美替代系统记事本。(239字)
|
12天前
|
人工智能 JSON 自然语言处理
让教学更智慧:用阿里云百炼工作流,自动生成中小学教材内容#小有可为#有温度的AI
通过可视化工作流编排,将大模型推理能力转化为标准化的教学内容生成引擎。教师只需输入教材标题和适用学段,即可自动获得结构完整、符合课程标准的章节内容,大幅降低备课门槛,助力教育资源均衡化。
493 127
|
6天前
|
人工智能 缓存 安全
Claude Code 封号真实原因曝光,这次彻底不装了,直接针对国内开发者的账号下手?
Claude Code 封号潮背后:逆向扒出客户端隐写区域标记,Anthropic 政策收紧叠加 DeepSeek 7 月涨价,国产替代更紧迫。
|
16天前
|
存储 人工智能 监控
QoderWork完全指南:从入门到精通,把“AI实习生”变成你的全能工作搭档
阿里云2026年推出的桌面端AI工作助手QoderWork,不止聊天,更可动手干活:本地运行、安全可控,支持文件整理、数据分析、PPT生成、网页开发等;内置专家套件、多Agent协作与自定义Skills,让AI真正成为你身边的“AI实习生”。
|
7天前
|
人工智能 安全 程序员
终于,Claude Code 封号的原因被曝光了!竟然针对中国用户,植入隐形代码?!
通俗易懂地揭秘 Claude Code 封号的手段,分享一些自己对 AI 编程困境的思考,Codex、Cursor、DeepSeek、智谱 GLM、甚至是豆包,都有所行动了
422 1
|
8天前
|
人工智能 安全 Cloud Native
Higress 新发布:AI Gateway 能力增强,Gateway API 及其推理扩展持续打磨
增强 AI 网关能力,持续打磨 Gateway API 及其推理扩展。
372 125
|
16天前
|
机器学习/深度学习 人工智能 调度
🐴 HappyHorse 1.1 现已上线阿里云百炼!快来查收模型使用指南,现在调用享 6 折~
HappyHorse 1.1 是新一代视频生成大模型,全面升级动态表现力、角色一致性、指令遵循、视觉质感与音画协同能力。支持I2V/T2V/R2V三类生成,适配短剧、电商广告、品牌营销等场景,提供高质、流畅、可控的AI视频生产力。
924 5
🐴 HappyHorse 1.1 现已上线阿里云百炼!快来查收模型使用指南,现在调用享 6 折~