Selenium Grid + CaptchaAI：分布式验证码求解

把验证码识别接进 Selenium Grid，卡住吞吐的通常不是浏览器不够多，而是两个上限没对齐：Grid 给多少个会话槽位，CaptchaAI 能同时处理多少个任务。“加了节点却没变快”多半出在这里。

下面走完整条链路：起 Grid 4、调 in.php / res.php 识别 reCAPTCHA v2 与 Turnstile、按空闲槽位定线程池，再交给 Kubernetes 扩缩。

Selenium Grid 验证码识别：先对齐两个并发上限

Grid 的上限是槽位：SE_NODE_MAX_SESSIONS 决定单节点跑几个 Chrome，3 节点各 5 个就是 15 个会话。CaptchaAI 的上限是线程：一个线程 = 一个在处理中的任务，按并发线程计费，每线程当月识别次数不限（STANDARD $30 / 15 线程，ADVANCE $90 / 50 线程）。

两个数不必相等：会话只有在等结果那几十秒占线程。

架构：多节点共用一个 API Key

┌─────────────┐     ┌──────────────┐     ┌──────────────┐
│  Test Script │────▶│  Grid Hub    │────▶│  Node 1      │
│  (Client)    │     │  (Router)    │     │  Chrome x 5  │
└─────────────┘     └──────────────┘     └──────────────┘
                           │              ┌──────────────┐
                           ├─────────────▶│  Node 2      │
                           │              │  Chrome x 5  │
                           │              └──────────────┘
                           │              ┌──────────────┐
                           └─────────────▶│  Node 3      │
                                          │  Chrome x 5  │
                                          └──────────────┘

All nodes share ──▶ CaptchaAI API (single API key)

关键是最后一行：所有节点共用一个 API Key。节点只管开浏览器、取 sitekey、把 token 写回表单。

用 Docker Compose 起 Grid 4

一个 Hub 加三个 Chrome 节点，每节点 5 个槽位：

version: "3"
services:
  selenium-hub:
    image: selenium/hub:4.21.0
    container_name: selenium-hub
    ports:

      - "4442:4442"
      - "4443:4443"
      - "4444:4444"

  chrome-node-1:
    image: selenium/node-chrome:4.21.0
    depends_on:

      - selenium-hub
    environment:

      - SE_EVENT_BUS_HOST=selenium-hub
      - SE_EVENT_BUS_PUBLISH_PORT=4442
      - SE_EVENT_BUS_SUBSCRIBE_PORT=4443
      - SE_NODE_MAX_SESSIONS=5
      - SE_NODE_OVERRIDE_MAX_SESSIONS=true

  chrome-node-2:
    image: selenium/node-chrome:4.21.0
    depends_on:

      - selenium-hub
    environment:

      - SE_EVENT_BUS_HOST=selenium-hub
      - SE_EVENT_BUS_PUBLISH_PORT=4442
      - SE_EVENT_BUS_SUBSCRIBE_PORT=4443
      - SE_NODE_MAX_SESSIONS=5
      - SE_NODE_OVERRIDE_MAX_SESSIONS=true

  chrome-node-3:
    image: selenium/node-chrome:4.21.0
    depends_on:

      - selenium-hub
    environment:

      - SE_EVENT_BUS_HOST=selenium-hub
      - SE_EVENT_BUS_PUBLISH_PORT=4442
      - SE_EVENT_BUS_SUBSCRIBE_PORT=4443
      - SE_NODE_MAX_SESSIONS=5
      - SE_NODE_OVERRIDE_MAX_SESSIONS=true

docker-compose up -d

打开 http://localhost:4444 看到三个节点各 5 个槽位即就绪。国内机器拉 selenium/node-chrome 镜像慢是常事，提前配好镜像加速。

在节点里调 CaptchaAI 识别 reCAPTCHA v2

这个类合并了两件事：申请 Grid 远程会话，以及提交并轮询识别任务。提交走 in.php——reCAPTCHA v2 用 method=userrecaptcha 传 googlekey，Turnstile 用 method=turnstile 传 sitekey；结果在 res.php 每 5 秒轮询，拿到 token 写回 g-recaptcha-response。

import requests
import time
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from concurrent.futures import ThreadPoolExecutor, as_completed


class GridCaptchaSolver:
    CAPTCHAAI_URL = "https://ocr.captchaai.com"

    def __init__(self, api_key, grid_url="http://localhost:4444"):
        self.api_key = api_key
        self.grid_url = grid_url

    def create_session(self):
        """Create a new browser session on the Grid."""
        options = webdriver.ChromeOptions()
        options.add_argument("--no-sandbox")
        options.add_argument("--no-sandbox")
        options.add_argument("--window-size=1920,1080")

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

    def solve_recaptcha_v2(self, site_url, sitekey):
        """Solve reCAPTCHA v2 via CaptchaAI API."""
        # Submit
        resp = requests.post(f"{self.CAPTCHAAI_URL}/in.php", data={
            "key": self.api_key,
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": site_url,
            "json": 1,
        })
        data = resp.json()
        if data["status"] != 1:
            raise Exception(f"Submit: {data['request']}")

        task_id = data["request"]

        # Poll
        for _ in range(60):
            time.sleep(5)
            resp = requests.get(f"{self.CAPTCHAAI_URL}/res.php", params={
                "key": self.api_key, "action": "get",
                "id": task_id, "json": 1,
            })
            data = resp.json()
            if data["request"] == "CAPCHA_NOT_READY":
                continue
            if data["status"] != 1:
                raise Exception(f"Solve: {data['request']}")
            return data["request"]

        raise Exception("Timeout")

    def solve_turnstile(self, site_url, sitekey):
        resp = requests.post(f"{self.CAPTCHAAI_URL}/in.php", data={
            "key": self.api_key, "method": "turnstile",
            "sitekey": sitekey, "pageurl": site_url, "json": 1,
        })
        data = resp.json()
        if data["status"] != 1:
            raise Exception(f"Submit: {data['request']}")

        task_id = data["request"]
        for _ in range(60):
            time.sleep(5)
            resp = requests.get(f"{self.CAPTCHAAI_URL}/res.php", params={
                "key": self.api_key, "action": "get",
                "id": task_id, "json": 1,
            })
            data = resp.json()
            if data["request"] == "CAPCHA_NOT_READY":
                continue
            if data["status"] != 1:
                raise Exception(f"Solve: {data['request']}")
            return data["request"]

        raise Exception("Timeout")

    def process_task(self, task):
        """Process a single CAPTCHA-protected task on a Grid node."""
        driver = self.create_session()

        try:
            driver.get(task["url"])
            time.sleep(2)

            # Detect sitekey
            sitekey = task.get("sitekey")
            if not sitekey:
                sitekey = driver.execute_script(
                    "return document.querySelector('[data-sitekey]')?.getAttribute('data-sitekey')"
                )

            if not sitekey:
                return {"url": task["url"], "status": "no_captcha", "data": driver.page_source[:500]}

            # Solve
            token = self.solve_recaptcha_v2(task["url"], sitekey)

            # Inject
            driver.execute_script(f"""
                document.querySelector('#g-recaptcha-response').value = '{token}';
                document.querySelectorAll('[name="g-recaptcha-response"]').forEach(
                    el => el.value = '{token}'
                );
            """)

            # Fill form and submit
            if task.get("form_data"):
                for field, value in task["form_data"].items():
                    driver.find_element(By.NAME, field).send_keys(value)

            if task.get("submit_selector"):
                driver.find_element(By.CSS_SELECTOR, task["submit_selector"]).click()
                time.sleep(3)

            return {
                "url": task["url"],
                "status": "success",
                "result_url": driver.current_url,
                "data": driver.page_source[:1000],
            }

        except Exception as e:
            return {"url": task["url"], "status": "error", "error": str(e)}

        finally:
            driver.quit()

finally 里的 driver.quit() 不能省：槽位不会自动回收，漏掉它跑几十个任务后就报 SessionNotCreated。

并行分发：线程池开多大

max_workers 是同时在跑的会话数，必须小于等于 Grid 总槽位，否则多出来的任务只是堵在 Hub 队列里——看着像变慢，其实没排上。

def run_parallel_tasks(api_key, tasks, max_workers=10):
    """Run CAPTCHA tasks in parallel across Grid nodes."""
    solver = GridCaptchaSolver(api_key)
    results = []

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(solver.process_task, task): task
            for task in tasks
        }

        for future in as_completed(futures):
            task = futures[future]
            try:
                result = future.result(timeout=600)
                results.append(result)
                print(f"[{result['status']}] {result['url']}")
            except Exception as e:
                results.append({
                    "url": task["url"],
                    "status": "exception",
                    "error": str(e),
                })

    return results


# Usage
tasks = [
    {
        "url": "https://site-a.com/form",
        "sitekey": "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
        "form_data": {"name": "Test User", "email": "[email protected]"},
        "submit_selector": "#submit",
    },
    {
        "url": "https://site-b.com/register",
        "sitekey": "6LdKlZEpAAAAAAOQjzC2v_mJ-",
        "form_data": {"username": "testuser"},
        "submit_selector": "button[type='submit']",
    },
    # Add more tasks...
]

results = run_parallel_tasks("YOUR_API_KEY", tasks, max_workers=15)

# Summary
success = sum(1 for r in results if r["status"] == "success")
print(f"\nCompleted: {success}/{len(results)} successful")

按空闲槽位动态决定识别并发

把 max_workers 写死只在节点数固定时成立。节点会变时，更稳的是每批开跑前问一次 /status：

import requests

def check_grid_status(grid_url="http://localhost:4444"):
    """Check Selenium Grid status and available nodes."""
    try:
        resp = requests.get(f"{grid_url}/status")
        data = resp.json()

        nodes = data.get("value", {}).get("nodes", [])
        total_slots = 0
        available_slots = 0

        print(f"Grid Status: {data['value']['ready']}")
        print(f"Nodes: {len(nodes)}")

        for i, node in enumerate(nodes):
            slots = node.get("slots", [])
            free = sum(1 for s in slots if not s.get("session"))
            total_slots += len(slots)
            available_slots += free
            print(f"  Node {i+1}: {free}/{len(slots)} slots available")

        print(f"Total capacity: {available_slots}/{total_slots} available")
        return available_slots

    except Exception as e:
        print(f"Grid check failed: {e}")
        return 0


# Adjust workers based on grid capacity
available = check_grid_status()
optimal_workers = min(available, 20)
print(f"Optimal workers: {optimal_workers}")

把 available_slots 和轮询轮数一起打进监控，扩容方向就不用猜：槽位富余而轮询在涨就加线程，槽位一直占满就加节点。

用 Kubernetes HPA 自动扩缩节点

任务量按天波动时，让节点跟着 CPU 走比养一堆空闲 Chrome 划算：

# selenium-grid-k8s.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: selenium-chrome-node
spec:
  replicas: 5
  selector:
    matchLabels:
      app: selenium-chrome
  template:
    metadata:
      labels:
        app: selenium-chrome
    spec:
      containers:

        - name: chrome
          image: selenium/node-chrome:4.21.0
          env:

            - name: SE_EVENT_BUS_HOST
              value: selenium-hub

            - name: SE_EVENT_BUS_PUBLISH_PORT
              value: "4442"

            - name: SE_EVENT_BUS_SUBSCRIBE_PORT
              value: "4443"

            - name: SE_NODE_MAX_SESSIONS
              value: "3"
          resources:
            limits:
              memory: "2Gi"
              cpu: "1"
            requests:
              memory: "1Gi"
              cpu: "500m"
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: chrome-node-hpa
spec:
  scaleRef:
    apiVersion: apps/v1
    kind: Deployment
    name: selenium-chrome-node
  minReplicas: 2
  maxReplicas: 20
  metrics:

    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 70

SE_NODE_MAX_SESSIONS 降到 3，是因为 Pod 只有 2 GiB 内存，会话开多了会被 OOM，而 Pod 一挂就连带打断它上面所有会话。20 个副本对应 60 个槽位，扩之前把线程数一并算好。

Java 侧的等价写法

用 TestNG 或 JUnit 跑集成测试时，客户端一般也留在 Java 里，结构与 Python 版一致。

import org.openqa.selenium.WebDriver;
import org.openqa.selenium.remote.RemoteWebDriver;
import org.openqa.selenium.chrome.ChromeOptions;
import java.net.URL;
import java.net.http.*;
import java.net.URI;
import java.util.concurrent.*;

public class GridCaptchaSolver {
    private final String apiKey;
    private final String gridUrl;
    private final HttpClient httpClient;

    public GridCaptchaSolver(String apiKey, String gridUrl) {
        this.apiKey = apiKey;
        this.gridUrl = gridUrl;
        this.httpClient = HttpClient.newHttpClient();
    }

    public WebDriver createSession() throws Exception {
        ChromeOptions options = new ChromeOptions();
        options.addArguments("--no-sandbox", "--window-size=1920,1080");
        return new RemoteWebDriver(new URL(gridUrl), options);
    }

    public List<Map<String, String>> runParallel(
        List<Map<String, String>> tasks, int workers
    ) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(workers);
        List<Future<Map<String, String>>> futures = new ArrayList<>();

        for (Map<String, String> task : tasks) {
            futures.add(executor.submit(() -> processTask(task)));
        }

        List<Map<String, String>> results = new ArrayList<>();
        for (Future<Map<String, String>> future : futures) {
            results.add(future.get(600, TimeUnit.SECONDS));
        }

        executor.shutdown();
        return results;
    }
}

国内环境的两个坑

一是页面没渲染出验证码控件。 reCAPTCHA 依赖 Google 托管的脚本，内地网络下不一定能稳定加载；脚本没起来时 [data-sitekey] 取不到值，任务以 no_captcha 收尾。先打印 driver.page_source 判断，别当成识别失败去重试。

二是目标站点用的不是 reCAPTCHA。 国内站点大量使用 GeeTest（极验）、网易易盾、腾讯防水墙，与国际站点常见的 reCAPTCHA / Turnstile 不是一套；CaptchaAI 覆盖 reCAPTCHA v2/v3、Turnstile、Cloudflare Challenge、GeeTest v3、图片与九宫格验证码和 BLS。另外只对自有或已授权的站点做自动化，并遵守 robots 协议。

排错清单

常见问题

每个节点开多少并发合适？

按内存倒推：一个 Chrome 会话约 1 GB，SE_NODE_MAX_SESSIONS 一般落在 3–5，实跑再按 /status 的空闲槽位取值。

线程数要和 Grid 并发配成 1:1 吗？

不用。会话只有等结果时占线程，加载、填表、提交都不占，先 1:1 起步再往下压。

所有节点共用一个 API Key 会互相影响吗？

不会。任务在 CaptchaAI 侧统一排队，共享的上限只有套餐线程数。

哪些验证码类型 CaptchaAI 还不支持？

hCaptcha 与 FunCaptcha（Arkose Labs）不在覆盖范围内。GeeTest v4 目前也不可用，官方口径是即将支持，能用的是 GeeTest v3。

相关指南

• Python + Selenium 集成 CaptchaAI
• Selenium Wire 请求拦截
• Node.js 工作线程并行识别

领取 CaptchaAI API Key，把验证码识别铺到整组 Grid 节点上。