批量验证码识别：并发提交与轮询实战

1000 个页面的验证码，一个个顺序解决要多久？按每个 15 秒算，将近 4 个小时。把提交和轮询拆开并发处理，同样的 1000 个任务几分钟就能跑完。这篇教程基于 CaptchaAI 的 in.php / res.php 接口，给出三种批量识别方案的 Python 实现，并说明并发数该怎么和套餐线程数对齐。

顺序等待为什么拖慢批量识别

顺序模式慢，卡在两步：

每个任务都要等上一个的轮询结果返回，才能提交下一个
等待时间是所有任务耗时的累加，不是并行时的最大值

并行提交、统一轮询，总耗时约等于耗时最长的那一个任务。

Sequential (slow):
  Submit #1 → Poll → Result (15s)
  Submit #2 → Poll → Result (15s)
  Submit #3 → Poll → Result (15s)
  Total: ~45s for 3 solves

Parallel (fast):
  Submit #1 ─┐
  Submit #2 ─┤→ Poll all → Results arrive
  Submit #3 ─┘
  Total: ~15s for 3 solves

下面依次给出线程池、asyncio、"先提交后轮询"三种实现。

方案一：线程池并发批量识别

最简单的方式是标准库自带的 ThreadPoolExecutor，无需额外依赖。submit_task 提交任务拿 task_id，poll_result 每 5 秒轮询直到不再是 CAPCHA_NOT_READY，batch_solve 用 max_workers 控制并发——设成 5–10 通常够用。

import requests
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"


def submit_task(method, **params):
    """Submit a single CAPTCHA task."""
    data = {"key": API_KEY, "method": method, "json": 1}
    data.update(params)
    resp = requests.post(f"{BASE_URL}/in.php", data=data, timeout=30)
    result = resp.json()
    if result.get("status") != 1:
        raise RuntimeError(f"Submit error: {result.get('request')}")
    return result["request"]


def poll_result(task_id, timeout=120):
    """Poll until result is ready."""
    start = time.time()
    while time.time() - start < timeout:
        time.sleep(5)
        resp = requests.get(f"{BASE_URL}/res.php", params={
            "key": API_KEY, "action": "get",
            "id": task_id, "json": 1,
        }, timeout=15)
        data = resp.json()
        if data["request"] != "CAPCHA_NOT_READY":
            return data["request"]
    raise TimeoutError(f"Task {task_id} timeout")


def solve_one(sitekey, pageurl):
    """Submit and poll a single task."""
    task_id = submit_task("userrecaptcha", googlekey=sitekey, pageurl=pageurl)
    token = poll_result(task_id)
    return {"url": pageurl, "token": token}


def batch_solve(tasks, max_workers=10):
    """Solve multiple CAPTCHAs in parallel."""
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(solve_one, t["sitekey"], t["url"]): t
            for t in tasks
        }
        for future in as_completed(futures):
            task = futures[future]
            try:
                result = future.result()
                results.append(result)
                print(f"Solved: {result['url']}")
            except Exception as e:
                print(f"Failed: {task['url']} - {e}")
                results.append({"url": task["url"], "token": None, "error": str(e)})

    return results


# Usage
tasks = [
    {"sitekey": "SITE_KEY_1", "url": "https://example.com/page1"},
    {"sitekey": "SITE_KEY_2", "url": "https://example.com/page2"},
    {"sitekey": "SITE_KEY_3", "url": "https://example.com/page3"},
]

results = batch_solve(tasks, max_workers=5)
print(f"Solved {sum(1 for r in results if r.get('token'))}/{len(tasks)}")

results 里既有成功拿到 token 的记录，也有失败原因，方便重试或记日志。

方案二：asyncio 异步批量识别，应对更高并发

线程池对几十并发够用，但几百线程同时跑，上下文切换会变成新瓶颈。100+ 并发时 asyncio + aiohttp 更划算，内存占用更低。下面用 Semaphore 控制 max_concurrent，避免打光接口。

import asyncio
import aiohttp
import time

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"


async def submit_task_async(session, method, **params):
    data = {"key": API_KEY, "method": method, "json": 1}
    data.update(params)
    async with session.post(f"{BASE_URL}/in.php", data=data) as resp:
        result = await resp.json()
        if result.get("status") != 1:
            raise RuntimeError(f"Submit error: {result.get('request')}")
        return result["request"]


async def poll_result_async(session, task_id, timeout=120):
    start = time.time()
    while time.time() - start < timeout:
        await asyncio.sleep(5)
        params = {
            "key": API_KEY, "action": "get",
            "id": task_id, "json": 1,
        }
        async with session.get(f"{BASE_URL}/res.php", params=params) as resp:
            data = await resp.json()
            if data["request"] != "CAPCHA_NOT_READY":
                return data["request"]
    raise TimeoutError(f"Task {task_id} timeout")


async def solve_one_async(session, sitekey, pageurl):
    task_id = await submit_task_async(
        session, "userrecaptcha",
        googlekey=sitekey, pageurl=pageurl,
    )
    token = await poll_result_async(session, task_id)
    return {"url": pageurl, "token": token}


async def batch_solve_async(tasks, max_concurrent=20):
    """Solve many CAPTCHAs concurrently with asyncio."""
    semaphore = asyncio.Semaphore(max_concurrent)
    results = []

    async def solve_with_limit(task):
        async with semaphore:
            try:
                result = await solve_one_async(
                    session, task["sitekey"], task["url"],
                )
                return result
            except Exception as e:
                return {"url": task["url"], "token": None, "error": str(e)}

    async with aiohttp.ClientSession() as session:
        coros = [solve_with_limit(t) for t in tasks]
        results = await asyncio.gather(*coros)

    return results


# Usage
tasks = [
    {"sitekey": "KEY", "url": f"https://example.com/page{i}"}
    for i in range(50)
]

results = asyncio.run(batch_solve_async(tasks, max_concurrent=20))
solved = sum(1 for r in results if r.get("token"))
print(f"Solved: {solved}/{len(tasks)}")

提交、轮询、限流都封装进了 solve_with_limit，几百个验证码可以在一次 asyncio.run 里跑完。

方案三：先提交后轮询，把吞吐量拉到最大

前两种方案提交后立刻在同一协程/线程里等结果，提交和轮询绑在一起。想要更高吞吐量，可以把两步拆开：先提交所有任务拿到全部 task_id，再统一轮询直到都返回或超时，适合任务量固定的场景。

import requests
import time

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"


def batch_submit(tasks):
    """Submit all tasks first, return task IDs."""
    submitted = []
    for task in tasks:
        try:
            data = {
                "key": API_KEY,
                "method": "userrecaptcha",
                "googlekey": task["sitekey"],
                "pageurl": task["url"],
                "json": 1,
            }
            resp = requests.post(f"{BASE_URL}/in.php", data=data, timeout=30)
            result = resp.json()
            if result.get("status") == 1:
                submitted.append({
                    "task_id": result["request"],
                    "url": task["url"],
                })
            time.sleep(0.1)  # Brief delay between submits
        except Exception as e:
            print(f"Submit failed for {task['url']}: {e}")
    return submitted


def batch_poll(submitted, timeout=120):
    """Poll all submitted tasks until complete."""
    pending = {s["task_id"]: s for s in submitted}
    results = []
    start = time.time()

    while pending and time.time() - start < timeout:
        time.sleep(5)
        for task_id in list(pending.keys()):
            try:
                resp = requests.get(f"{BASE_URL}/res.php", params={
                    "key": API_KEY, "action": "get",
                    "id": task_id, "json": 1,
                }, timeout=15)
                data = resp.json()
                if data["request"] != "CAPCHA_NOT_READY":
                    info = pending.pop(task_id)
                    results.append({
                        "url": info["url"],
                        "token": data["request"],
                    })
            except Exception:
                pass

    # Mark remaining as failed
    for task_id, info in pending.items():
        results.append({"url": info["url"], "token": None, "error": "timeout"})

    return results


# Usage
tasks = [
    {"sitekey": "KEY", "url": f"https://example.com/page{i}"}
    for i in range(20)
]

submitted = batch_submit(tasks)
print(f"Submitted {len(submitted)} tasks")

results = batch_poll(submitted)
solved = sum(1 for r in results if r.get("token"))
print(f"Solved: {solved}/{len(tasks)}")

batch_submit 每次提交间留 100 毫秒间隔避免限流；batch_poll 用 pending 字典跟踪未完成任务，超时的会被标记失败，不拖住整个批次。

线程数与套餐档位如何匹配

max_workers 超过套餐线程上限时，多出的任务只会排队，不会更快——CaptchaAI 按并发线程数计费，不按验证码数量收费。

常见档位：

BASIC —— $15/月，5 线程
STANDARD —— $30/月，15 线程
ADVANCE —— $90/月，50 线程
PREMIUM —— $170/月，100 线程
CORPORATE —— $240/月，150 线程
ENTERPRISE —— $300/月，200 线程
VIP-1 至 VIP-3：1,000–5,000 线程，大规模采集用

吞吐量参考

大致速度参考（实际因站点响应速度、验证码类型和网络状况波动）：

并发任务数	大致速度	适用场景
1–5	3–5 次/分钟	测试、小规模抓取
5–20	15–60 次/分钟	生产环境抓取
20–50	60–150 次/分钟	高流量数据管道
50–100	150–300 次/分钟	企业级规模

常见故障排查

问题	原因	处理方式
频繁触发限流（429）	单位时间内提交次数过多	每次提交之间加 100 毫秒左右的间隔
大量任务超时	轮询的 timeout 设置太短	把 timeout 调到 120–180 秒
并发数超过 50 后提速不明显	网络请求本身成为瓶颈	换成 asyncio + aiohttp，而不是继续加线程
结果对应错乱	用列表而不是字典跟踪 task_id	统一用以 task_id 为 key 的字典保存结果

常见问题

套餐线程数和脚本里的并发数是什么关系？

STANDARD 有 15 个线程，max_workers 设成 10–15 刚好用满；设成 50 不会更快三倍，只会排队等线程空出来。

线程池还是 asyncio，人少的爬虫项目该怎么选？

几十并发用 ThreadPoolExecutor 更直观；100+ 并发时 asyncio 配合 aiohttp 资源占用更低，但协程调试门槛更高。

批量提交后经常超时或部分失败，该怎么排查？

先确认 timeout 给够 120 秒，再看是不是提交过密触发限流；调大间隔、拉长 timeout 通常就能解决。

GeeTest、Turnstile 也能用同样的批量模式吗？

可以，把 method 换成对应类型（Turnstile 用 turnstile，GeeTest v3 用 geetest）即可，逻辑不变。国内站点更常见 GeeTest（极验）而不是 reCAPTCHA。

批量验证码求解：高效提交多个任务

顺序等待为什么拖慢批量识别

方案一：线程池并发批量识别

方案二：asyncio 异步批量识别，应对更高并发

方案三：先提交后轮询，把吞吐量拉到最大

线程数与套餐档位如何匹配

吞吐量参考

常见故障排查

常见问题

套餐线程数和脚本里的并发数是什么关系？

线程池还是 asyncio，人少的爬虫项目该怎么选？

批量提交后经常超时或部分失败，该怎么排查？

GeeTest、Turnstile 也能用同样的批量模式吗？

相关指南

AWS Lambda + CaptchaAI：无服务器验证码解决

用于 CaptchaAI Worker 部署的 Ansible Playbook

使用 AWS SNS 和 CaptchaAI 构建事件驱动的验证码解决方案

验证码解决基础设施的蓝绿部署

Azure Functions + CaptchaAI：云集成

构建 CaptchaAI 使用仪表板和监控

顺序等待为什么拖慢批量识别

方案一：线程池并发批量识别

方案二：asyncio 异步批量识别，应对更高并发

方案三：先提交后轮询，把吞吐量拉到最大

线程数与套餐档位如何匹配

吞吐量参考

常见故障排查

常见问题

套餐线程数和脚本里的并发数是什么关系？

线程池还是 asyncio，人少的爬虫项目该怎么选？

批量提交后经常超时或部分失败，该怎么排查？

GeeTest、Turnstile 也能用同样的批量模式吗？

相关指南

相关文章

AWS Lambda + CaptchaAI：无服务器验证码解决

用于 CaptchaAI Worker 部署的 Ansible Playbook

使用 AWS SNS 和 CaptchaAI 构建事件驱动的验证码解决方案

验证码解决基础设施的蓝绿部署

Azure Functions + CaptchaAI：云集成

构建 CaptchaAI 使用仪表板和监控