如何用 Python asyncio 并行识别验证码

8 月 11, 2026 已关闭评论

solve captchas in parallel - How to Solve CAPTCHAs in Parallel with Python asyncio

如果你需要在 Python 中并行识别验证码，请使用 AsyncCapSkip 连同 asyncio.gather。Python 是唯一一个异步客户端为真正异步实现、而不是别名的 CapSkip SDK，所以十个 reCAPTCHA 的批量任务大致只花最慢那一个的时间，而不是十个之和。本文覆盖可用的代码、如何限制并发以免压垮识别工具，以及如何避免一次失败毁掉整批。

为什么 Python 客户端才是关键

四个 SDK 都导出了一个叫做 AsyncCapSkip的东西。但只有一个是独立实现。

SDK	What `AsyncCapSkip` is	如何并发地跑任务
Python	真正的异步客户端	`await asyncio.gather(...)`
Node.js	别名，指向 `CapSkip`	`await Promise.all([...])`
.NET	别名，指向 `CapSkipClient`	`await Task.WhenAll(...)`
PHP	仅仅是别名，用于保持源码一致	同步执行，没有并发

Node 和 .NET 本身就是非阻塞的，所以别名在那里没有任何代价。PHP 是同步的，别名对你毫无帮助。在 Python 里这个区别很重要：普通的 CapSkip 客户端在轮询时会阻塞事件循环，所以把它放进协程只是名义上的并发，实际耗时依然是串行的。

你需要什么

Python 3.10 或更高版本
pip install capskip
CapSkip 正在运行且已开启 API 服务，监听于 127.0.0.1:8080
一份待处理的 sitekey 与页面 URL 列表

没有任何数据离开你的机器，所以不需要迁就速率限制，做实验时也没有按次计费在跑。每种类型的完整方法签名都在 CAPTCHA 识别 SDK 页面。

串行写法，以及它的代价

大多数人一开始都会这样写。它是对的，但很慢。

# pip install capskip
from capskip import CapSkip

solver = CapSkip(host="127.0.0.1", port=8080)

targets = [
    ("SITEKEY_A", "https://example.com/page-a"),
    ("SITEKEY_B", "https://example.com/page-b"),
    ("SITEKEY_C", "https://example.com/page-c"),
]

# Each call blocks until that one CAPTCHA comes back.
for sitekey, url in targets:
    result = solver.recaptcha(sitekey=sitekey, url=url)
    print(result["code"][:40])   # the token, truncated for the log

reCAPTCHA 的识别过程大部分时间都在等待。识别工具干活时你的进程闲着，然后换下一个，接着又闲着。三次识别就要花三次识别的墙钟时间。三十次就是三十次。

用 asyncio.gather 一次识别多个验证码

换掉客户端，await 这些调用，然后把它们一起交给 gather。类型可以混着来：reCAPTCHA、Turnstile 和极验放在同一批里没有问题。

# pip install capskip
import asyncio
from capskip import AsyncCapSkip

async def main():
    solver = AsyncCapSkip(host="127.0.0.1", port=8080)

    # gather starts all three now and waits for the slowest.
    results = await asyncio.gather(
        solver.recaptcha(sitekey="SITEKEY_A", url="https://example.com/page-a"),
        solver.turnstile(sitekey="SITEKEY_B", url="https://example.com/page-b"),
        solver.normal("captcha.png"),
    )

    for r in results:
        print(r["code"][:40])   # token for widgets, text for images

asyncio.run(main())

每个方法都返回一个带有相同核心字段的字典： captchaId 和 code。Turnstile 还会多返回一个 userAgent，提交挑战页的识别结果时必须连同 token 一起回传。极验则会多返回 challenge, validate 和 seccode，并且把原始 JSON 放在 code.

用信号量限制并发数

不要一口气朝本地守护进程扔两百个识别任务碰运气。识别是跑在你自己机器上的 CPU 工作，超过某个宽度之后你只是在跟自己排队，每一次识别都会更慢。信号量能把同时进行的数量固定住。

import asyncio
from capskip import AsyncCapSkip

# Start at 4 or 5, then measure. More is not automatically faster.
sem = asyncio.Semaphore(5)

async def solve_one(solver, sitekey, url):
    async with sem:
        return await solver.recaptcha(sitekey=sitekey, url=url)

async def run(targets):
    solver = AsyncCapSkip()
    tasks = [solve_one(solver, k, u) for k, u in targets]
    return await asyncio.gather(*tasks)

用计时来选这个数字，别靠猜。把同样的 20 个目标分别按 2、5、10 跑一遍，留下在你硬件上最快的那个。正确答案取决于你的 CPU，而不是 SDK。

共用一个客户端，而不是每个任务一个

只创建一个 AsyncCapSkip ，并像上面那样在协程之间共用。每个任务都新建一个既浪费又没有好处：客户端保存的是配置，不是每次识别的状态。

别让一次失败毁掉整批

默认情况下， gather 会抛出第一个异常，于是其他正在进行的结果全部丢失。传入 return_exceptions=True 之后，异常会作为普通元素出现在结果列表里，你就能把成功的和失败的分开处理。

from capskip import (
    AsyncCapSkip, ApiException, NetworkException,
    TimeoutException, ValidationException,
)

results = await asyncio.gather(*tasks, return_exceptions=True)

for target, r in zip(targets, results):
    if isinstance(r, TimeoutException):
        print("timed out, worth retrying:", target)
    elif isinstance(r, ApiException):
        print("api rejected this one:", target, r)
    elif isinstance(r, NetworkException):
        print("solver unreachable, stop the run:", target)
    elif isinstance(r, Exception):
        raise r
    else:
        print("ok:", r["code"][:40])

四种异常类型在每个 CapSkip SDK 里都一样，并且全部继承自基类 CapSkipError ，如果你更愿意只捕获一种类型的话。 ValidationException 表示你的参数有问题，重试也会以同样的方式失败。 NetworkException 通常表示应用没有在运行，这是整轮任务的问题，而不是单个目标的问题。

超时与轮询：不同类型的行为并不一样

有两个独立的超时时间，混合类型的批次会同时受这两者约束。

选项	默认值	适用于
`defaultTimeout`	120 秒	图片验证码
`recaptchaTimeout`	300 秒	reCAPTCHA、Turnstile、极验
`pollingInterval`	5 秒	加载的最大轮询之间的间隔

pollingInterval 在调整它之前值得先弄明白。SDK 并不是按固定间隔轮询的。它从 250ms 起步，逐步退避到你设定的值，这也是为什么 SDK 的识别通常比照着原始 API「先等待，再每五秒轮询一次」的建议手写的循环返回得更快。调大它会让本来很快的识别更晚返回。调小它只会增加请求量，没有收益。

极验挑战会过期，所以不要预先攒一批

这一条恰恰是在你转向并行之后才会咬人。极验的 gt 值对每个站点是固定的，但 challenge 只能用一次，大约一分钟就过期。如果你先收集五十个挑战再开始识别，排在队尾的那些在提交之前就已经失效了。每个挑战都要在用到它的那次识别之前才去获取。

完整可运行示例

# pip install capskip
import asyncio
from capskip import AsyncCapSkip, ApiException, TimeoutException

TARGETS = [
    ("SITEKEY_A", "https://example.com/page-a"),
    ("SITEKEY_B", "https://example.com/page-b"),
    ("SITEKEY_C", "https://example.com/page-c"),
]

async def solve_one(solver, sem, sitekey, url):
    async with sem:
        return await solver.recaptcha(sitekey=sitekey, url=url)

async def main():
    solver = AsyncCapSkip(host="127.0.0.1", port=8080)
    sem = asyncio.Semaphore(5)

    tasks = [solve_one(solver, sem, k, u) for k, u in TARGETS]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    tokens = {}
    for (sitekey, url), r in zip(TARGETS, results):
        if isinstance(r, (ApiException, TimeoutException)):
            print("failed:", url, r)
        else:
            tokens[url] = r["code"]

    print(len(tokens), "of", len(TARGETS), "solved")
    return tokens

asyncio.run(main())

整个模式就是这样：一个共用的客户端、一个信号量、 return_exceptions=True，最后得到一个 token 字典。把它放进爬虫，验证码这一步就不再是瓶颈。网络爬虫验证码识别工具页面讲了它在更大的流程里的位置。

同样的思路在其他 SDK 里怎么写

如果你要移植这段代码，并发原语会变，但结构不变。Node 本身就是非阻塞的，所以普通客户端就够了。

// npm install capskip
const { CapSkip } = require('capskip');

const solver = new CapSkip({ host: '127.0.0.1', port: 8080 });

// AsyncCapSkip here is just an alias. Promise.all does the work.
const results = await Promise.all([
  solver.recaptcha('SITEKEY_A', 'https://example.com/page-a'),
  solver.turnstile('SITEKEY_B', 'https://example.com/page-b'),
]);

console.log(results.map(r => r.code));

.NET 也是一样，用 Task.WhenAll，而 PHP 根本没有并发可言。如果你需要并行识别，并且可以自己选语言，Python 是唯一带专门客户端的那个。 Python 验证码识别页面覆盖了其余的接口。

常见错误

错误做法	会发生什么	修复
使用 `CapSkip` （在协程内部）	阻塞事件循环，耗时依然是串行的	使用 `AsyncCapSkip`
没有信号量	队列一深，每一次识别都会变慢	限制同时进行的数量，从 5 左右开始
普通的 `gather`	一次失败会丢弃其他所有结果	`return_exceptions=True`
提前批量获取极验挑战	靠后的那些在提交前就过期了	每一个都在识别前才获取
调大 `pollingInterval`	本来很快的识别反而更晚返回	保持默认值
给图片识别设置了代理	图片验证码不支持代理	代理只适用于 reCAPTCHA、Turnstile 和极验

如果你想看看 SDK 到底发了什么，每种类型的原始请求与响应都在 API 文档。Python 官方的 asyncio 任务参考文档涵盖 gather 的语义有详细说明。

常见问题

我一次能识别多少个验证码？

没有配额可以撞上，所以上限就是你自己的硬件。识别在本地进行，因此并发受限于 CPU，而不是账户等级。先从同时 5 个开始，给一批固定任务计时，再据此调整。

可以在一次 gather 调用里混合不同的验证码类型吗？

可以。 recaptcha, turnstile, geetest 和 normal 都是同一个客户端上的协程，可以一起 await。记住图片识别用的是 120 秒超时，其余类型用 300 秒。

我应该改用线程吗？

只有当你周围的代码本来就是多线程时才需要。这里的工作是 I/O 等待，正是 asyncio 擅长的，而且一个事件循环比线程池更省。如果你被困在同步代码库里，用线程池包住普通的 CapSkip 客户端也可以。

AsyncCapSkip 需要关闭吗？

SDK 没有记载任何 close 方法或异步上下文管理器，所以创建一个客户端、在整轮任务里用它，进程结束时让它自然回收就行。

小结

使用 AsyncCapSkip，共用一个客户端，用信号量限制同时进行的识别数量，并传入 return_exceptions=True ，这样单个有问题的目标就不会毁掉整批。这能把一队验证码从串行瓶颈变成一次等待。

你之所以能放心加大并发，是因为识别工具跑在你自己的机器上。没有按次计费，也没有要和陌生人共享的队列，所以扩展并发只关乎你的 CPU，而不是别人的速率限制。当批量规模不再小的时候，本地验证码绕过工具带来的实际差别就在这里。

验证码类型

开发者

AI 集成

替代方案

快速开始

开发者资源