首页
学习
活动
专区
圈层
工具
发布
社区首页 >专栏 >2026年Selenium爬虫代理IP集成实战:从环境搭建到项目落地

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

原创
作者头像
LeoCrawls
发布2026-07-08 09:41:50
发布2026-07-08 09:41:50
100
举报

先说结论,再说为什么。

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

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

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

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

Python + Selenium 4.x + WebDriver Manager

代码语言:bash
复制
 # 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 的版本,省了一堆版本不对的报错。

代码语言:python
复制
 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 加了白名单。

代码语言:python
复制
 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 响应并自动填入账密。代码不长,但细节多,我直接给能跑的版本:

代码语言:python
复制
 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 轮换逻辑了。

代码语言:python
复制
 # 隧道代理集成(以账密认证为例)
 # 你拿到的隧道网关通常是一个固定地址,比如 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 代理不太一样。

代码语言:python
复制
 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。所以轮换逻辑本质上是一个"创建 → 使用 → 销毁 → 重建"的循环。

代码语言:python
复制
 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 月那单舆情监测项目的精简版,脱敏后拿出来。

代码语言:python
复制
 """
 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:

代码语言:python
复制
 # 安装 Xvfb(Ubuntu/Debian)
 sudo apt-get install xvfb
 ​
 # 用 Xvfb 启动你的脚本
 xvfb-run --auto-servernum python your_crawler.py

或者用 Python 的 pyvirtualdisplay

代码语言:python
复制
 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 模块加一层兜底:

代码语言:python
复制
 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 库做一次轻量级验活:

代码语言:python
复制
 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 缺失等。

代码语言:python
复制
 # 基础的反检测配置
 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 实例不能跨线程操作。所以多线程方案基本不用考虑,直接上多进程:

代码语言:python
复制
 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 节点跑实际的浏览器。

代码语言:python
复制
 # 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:

代码语言:python
复制
 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 轮换做好、异常处理兜住。框架在换,这三件事十年了没变过。

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 环境这一层,2026 年有几个变化
    • Python + Selenium 4.x + WebDriver Manager
  • 代理集成:三种方式,各有各的适用边界
    • 方式一:Chrome 启动参数直接注入(无认证代理)
    • 方式二:Chrome 插件注入认证信息(带账密的代理)
    • 方式三:隧道代理——最省事的方案
  • SOCKS5 代理的接法
  • IP 轮换:不用隧道代理时怎么自己做
  • 完整项目模板:舆情页面采集
  • 几个实际跑起来会遇到的坑
  • 并发采集的架构选择
  • 这些年的经验浓缩成几句话
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档