顺序求解很简单。并行求解速度很快。正确的选择取决于您的数量、基础设施和对复杂性的容忍度。本指南对这两种方法进行了比较CaptchaAI并帮助您选择正确的策略。
头对头比较
| 因素 | 顺序 | 平行线 |
|---|---|---|
| 吞吐量（reCAPTCHA v2，15秒中值） | ~240/hour | ~10,000+/hour |
| 代码复杂度 | 简单的 | 中度至复杂 |
| 错误处理 | 直截了当 | 需要并发错误隔离 |
| 内存使用情况 | 最小 (~30 MB) | 可随并发扩展 (~100–500 MB) |
| 每次解决的 API 成本 | 相同的 | 相同的 |
| 调试难度 | 简单的 | 更难（比赛条件、时间安排） |
| 订单保存 | 有保证 | 需要明确的跟踪 |
| 最适合 | < 500 解决/day | > 500 解决/day |
顺序求解
它是如何运作的
一次一个验证码：提交 → 等待 → 轮询 → 下一步得到结果 →。
# sequential_solver.py
import os
import time
import requests
API_KEY = os.environ.get("CAPTCHAAI_KEY", "YOUR_API_KEY")
def solve_sequential(tasks):
"""Solve CAPTCHAs one by one."""
results = []
session = requests.Session()
for task in tasks:
# Submit
resp = session.get("https://ocr.captchaai.com/in.php", params={
"key": API_KEY,
"method": "userrecaptcha",
"googlekey": task["sitekey"],
"pageurl": task["pageurl"],
"json": "1",
})
result = resp.json()
if result.get("status") != 1:
results.append({"error": result.get("request")})
continue
task_id = result["request"]
time.sleep(15)
# Poll
token = None
for _ in range(25):
poll = session.get("https://ocr.captchaai.com/res.php", params={
"key": API_KEY, "action": "get",
"id": task_id, "json": "1",
})
poll_result = poll.json()
if poll_result.get("status") == 1:
token = poll_result["request"]
break
if poll_result.get("request") != "CAPCHA_NOT_READY":
break
time.sleep(5)
results.append({"token": token} if token else {"error": "timeout"})
return results
# 10 tasks sequentially → ~150 seconds total
tasks = [{"sitekey": "SITEKEY", "pageurl": "https://example.com"}] * 10
start = time.time()
results = solve_sequential(tasks)
print(f"Completed in {time.time() - start:.0f}s")
当顺序有意义时
- 单页工作流程 — 一次填写一张表格
- 调试和开发 — 清晰的执行流程
- 低容量 — < 500 解决了/day 并不能证明并行复杂性是合理的
- 顺序相关任务 - 当求解结果必须按顺序使用时
并行求解
它是如何运作的
立即提交所有验证码，同时轮询结果：
# parallel_solver.py
import os
import asyncio
import aiohttp
API_KEY = os.environ.get("CAPTCHAAI_KEY", "YOUR_API_KEY")
async def solve_one(session, sitekey, pageurl, semaphore):
"""Solve a single CAPTCHA within concurrency limits."""
async with semaphore:
# Submit
async with session.get("https://ocr.captchaai.com/in.php", params={
"key": API_KEY, "method": "userrecaptcha",
"googlekey": sitekey, "pageurl": pageurl, "json": "1",
}) as resp:
result = await resp.json(content_type=None)
if result.get("status") != 1:
return {"error": result.get("request")}
task_id = result["request"]
await asyncio.sleep(15)
# Poll
for _ in range(25):
async with session.get("https://ocr.captchaai.com/res.php", params={
"key": API_KEY, "action": "get",
"id": task_id, "json": "1",
}) as resp:
poll_result = await resp.json(content_type=None)
if poll_result.get("status") == 1:
return {"token": poll_result["request"]}
if poll_result.get("request") != "CAPCHA_NOT_READY":
return {"error": poll_result.get("request")}
await asyncio.sleep(5)
return {"error": "timeout"}
async def solve_parallel(tasks, max_concurrent=50):
"""Solve CAPTCHAs in parallel with concurrency control."""
semaphore = asyncio.Semaphore(max_concurrent)
connector = aiohttp.TCPConnector(limit=max_concurrent)
async with aiohttp.ClientSession(connector=connector) as session:
coros = [
solve_one(session, t["sitekey"], t["pageurl"], semaphore)
for t in tasks
]
return await asyncio.gather(*coros)
# 10 tasks in parallel → ~20 seconds total
import time
tasks = [{"sitekey": "SITEKEY", "pageurl": "https://example.com"}] * 10
start = time.time()
results = asyncio.run(solve_parallel(tasks))
print(f"Completed in {time.time() - start:.0f}s")
JavaScript 并行示例
// parallel_solver.js
const axios = require('axios');
const https = require('https');
const API_KEY = process.env.CAPTCHAAI_KEY || 'YOUR_API_KEY';
const agent = new https.Agent({ keepAlive: true, maxSockets: 50 });
const api = axios.create({ baseURL: 'https://ocr.captchaai.com', httpsAgent: agent });
async function solveOne(sitekey, pageurl) {
const submit = await api.get('/in.php', {
params: { key: API_KEY, method: 'userrecaptcha', googlekey: sitekey, pageurl, json: '1' },
});
if (submit.data.status !== 1) return { error: submit.data.request };
await new Promise(r => setTimeout(r, 15000));
for (let i = 0; i < 25; i++) {
const poll = await api.get('/res.php', {
params: { key: API_KEY, action: 'get', id: submit.data.request, json: '1' },
});
if (poll.data.status === 1) return { token: poll.data.request };
if (poll.data.request !== 'CAPCHA_NOT_READY') return { error: poll.data.request };
await new Promise(r => setTimeout(r, 5000));
}
return { error: 'timeout' };
}
(async () => {
const tasks = Array.from({ length: 10 }, () => ({
sitekey: 'SITEKEY', pageurl: 'https://example.com',
}));
const start = Date.now();
const results = await Promise.all(tasks.map(t => solveOne(t.sitekey, t.pageurl)));
console.log(`Completed in ${((Date.now() - start) / 1000).toFixed(0)}s`);
console.log(`Solved: ${results.filter(r => r.token).length}/${tasks.length}`);
agent.destroy();
})();
按并发级别划分的吞吐量
| 并发解决 | 预计吞吐量/hour | 内存（Python） | 复杂 |
|---|---|---|---|
| 1（顺序） | 240 | 30MB | 低的 |
| 10 | 2,400 | 50MB | 低的 |
| 25 | 6,000 | 80MB | 中等的 |
| 50 | 10,000+ | 120MB | 中等的 |
| 100 | 18,000+ | 200MB | 高的 |
基于 reCAPTCHA v2，中值求解时间为 15 秒。
混合方法
对工作流程使用顺序，对验证码解决使用并行：
# Process 10 URLs sequentially, but solve their CAPTCHAs in a parallel batch
urls = get_next_batch() # 10 URLs
captcha_params = [extract_sitekey(url) for url in urls] # Sequential extraction
tokens = asyncio.run(solve_parallel(captcha_params, max_concurrent=10)) # Parallel solving
for url, result in zip(urls, tokens):
submit_form(url, result.get("token")) # Sequential submission
故障排除
| 问题 | 原因 | 处理方式 |
|---|---|---|
| 并行速度比预期慢 | 信号量限制太多 | 增加max_concurrent |
| 并行随机故障 | 共同的国家腐败 | 每个协程的隔离状态/promise |
| 结果乱序 | asyncio.gather 保留订单 |
如果需要，使用索引跟踪 |
API返回ERROR_NO_SLOT_AVAILABLE |
并发提交过多 | 在提交之间添加较小的延迟 |
常问问题
我可以在不改变代码结构的情况下从顺序切换到并行吗？
是的，如果您将求解逻辑提取到独立函数中。唯一的变化是用 asyncio.gather (Python) 或 Promise.all (JavaScript) 包装调用。
并行求解成本更高吗？
不会。CaptchaAI 按线程收费（每个线程不限解决次数），不按超出计划的并发请求收费。并行求解与顺序求解具有相同的 API 成本 — 您只是更快地得出总数。
最大实际并发数是多少？
大多数设置在超过 100 个并发求解时都会出现收益递减。除此之外，专注于优化求解参数而不是增加并发性。
下一步
选择与您的交易量相匹配的策略 —获取您的 CaptchaAI API 密钥。
相关指南：
- 并行验证码求解
- 每小时 10,000 个任务
- 内存和CPU优化
