在未经测试的情况下更换验证码提供商是有风险的。并行运行会同时向您当前的提供商和 CaptchaAI 发送相同的验证码挑战,为您提供有关解决率、速度和成本的并行数据。
为什么并行测试很重要
营销页面上的基准并不反映您的特定流量模式。您的验证码具有独特的特征 - 站点密钥、代理配置、地理分布。并行测试揭示了与您的实际工作负载的实际性能差异。
建筑学
┌──────────────┐
│ Your App │
└──────┬───────┘
│
┌──────▼───────┐
│ CAPTCHA │
│ Router │
└──┬───────┬───┘
│ │
┌────────▼──┐ ┌──▼────────┐
│ Current │ │ CaptchaAI │
│ Provider │ │ │
└────────┬──┘ └──┬────────┘
│ │
┌──▼───────▼──┐
│ Metrics │
│ Collector │
└─────────────┘
Python实现
提供者抽象
import os
import time
import requests
from dataclasses import dataclass, field
from typing import Optional
from concurrent.futures import ThreadPoolExecutor
@dataclass
class SolveResult:
provider: str
success: bool
solution: Optional[str] = None
error: Optional[str] = None
elapsed: float = 0.0
cost: float = 0.0
class CaptchaProvider:
def __init__(self, name, submit_url, result_url, api_key):
self.name = name
self.submit_url = submit_url
self.result_url = result_url
self.api_key = api_key
self.session = requests.Session()
def solve_recaptcha(self, sitekey, pageurl):
start = time.time()
resp = self.session.post(self.submit_url, data={
"key": self.api_key,
"method": "userrecaptcha",
"googlekey": sitekey,
"pageurl": pageurl,
"json": 1
})
data = resp.json()
if data.get("status") != 1:
return SolveResult(
provider=self.name, success=False,
error=data.get("request"), elapsed=time.time() - start
)
captcha_id = data["request"]
for _ in range(60):
time.sleep(5)
result = self.session.get(self.result_url, params={
"key": self.api_key, "action": "get",
"id": captcha_id, "json": 1
}).json()
if result.get("status") == 1:
return SolveResult(
provider=self.name, success=True,
solution=result["request"], elapsed=time.time() - start
)
if result.get("request") != "CAPCHA_NOT_READY":
return SolveResult(
provider=self.name, success=False,
error=result.get("request"), elapsed=time.time() - start
)
return SolveResult(
provider=self.name, success=False,
error="TIMEOUT", elapsed=time.time() - start
)
平行跑者
class ParallelTestRunner:
def __init__(self, primary, challenger):
self.primary = primary
self.challenger = challenger
self.results = {"primary": [], "challenger": []}
def run_test(self, sitekey, pageurl, num_runs=20):
print(f"Running {num_runs} parallel solves...")
for i in range(num_runs):
with ThreadPoolExecutor(max_workers=2) as executor:
primary_future = executor.submit(
self.primary.solve_recaptcha, sitekey, pageurl
)
challenger_future = executor.submit(
self.challenger.solve_recaptcha, sitekey, pageurl
)
primary_result = primary_future.result()
challenger_result = challenger_future.result()
self.results["primary"].append(primary_result)
self.results["challenger"].append(challenger_result)
print(f" Run {i+1}/{num_runs}: "
f"{self.primary.name}={'OK' if primary_result.success else 'FAIL'} "
f"({primary_result.elapsed:.1f}s) | "
f"{self.challenger.name}={'OK' if challenger_result.success else 'FAIL'} "
f"({challenger_result.elapsed:.1f}s)")
return self.generate_report()
def generate_report(self):
report = {}
for label, results in self.results.items():
total = len(results)
successes = sum(1 for r in results if r.success)
times = [r.elapsed for r in results if r.success]
errors = [r.error for r in results if not r.success]
report[label] = {
"provider": results[0].provider if results else "unknown",
"total": total,
"successes": successes,
"success_rate": (successes / total * 100) if total else 0,
"avg_time": sum(times) / len(times) if times else 0,
"min_time": min(times) if times else 0,
"max_time": max(times) if times else 0,
"errors": errors
}
return report
# Usage
current = CaptchaProvider(
name="CurrentProvider",
submit_url="https://current-provider.com/in.php",
result_url="https://current-provider.com/res.php",
api_key="current_key"
)
captchaai = CaptchaProvider(
name="CaptchaAI",
submit_url="https://ocr.captchaai.com/in.php",
result_url="https://ocr.captchaai.com/res.php",
api_key=os.environ["CAPTCHAAI_API_KEY"]
)
runner = ParallelTestRunner(primary=current, challenger=captchaai)
report = runner.run_test(
sitekey="6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
pageurl="https://example.com/form",
num_runs=20
)
for label, stats in report.items():
print(f"\n{stats['provider']}:")
print(f" Success rate: {stats['success_rate']:.1f}%")
print(f" Avg time: {stats['avg_time']:.1f}s")
print(f" Min/Max: {stats['min_time']:.1f}s / {stats['max_time']:.1f}s")
if stats['errors']:
print(f" Errors: {stats['errors']}")
流量分流
对于生产,将一定比例的流量路由到 CaptchaAI:
import random
class TrafficSplitter:
def __init__(self, primary, challenger, challenger_pct=10):
self.primary = primary
self.challenger = challenger
self.challenger_pct = challenger_pct
def solve(self, sitekey, pageurl):
if random.randint(1, 100) <= self.challenger_pct:
result = self.challenger.solve_recaptcha(sitekey, pageurl)
if not result.success:
# Fall back to primary on failure
return self.primary.solve_recaptcha(sitekey, pageurl)
return result
return self.primary.solve_recaptcha(sitekey, pageurl)
# Start with 10%, increase as confidence builds
splitter = TrafficSplitter(current, captchaai, challenger_pct=10)
result = splitter.solve(sitekey="...", pageurl="...")
JavaScript 实现
const axios = require("axios");
class CaptchaProvider {
constructor(name, submitUrl, resultUrl, apiKey) {
this.name = name;
this.submitUrl = submitUrl;
this.resultUrl = resultUrl;
this.apiKey = apiKey;
}
async solveRecaptcha(sitekey, pageurl) {
const start = Date.now();
try {
const submit = await axios.post(this.submitUrl, null, {
params: { key: this.apiKey, method: "userrecaptcha", googlekey: sitekey, pageurl, json: 1 },
});
if (submit.data.status !== 1) {
return { provider: this.name, success: false, error: submit.data.request, elapsed: (Date.now() - start) / 1000 };
}
const captchaId = submit.data.request;
for (let i = 0; i < 60; i++) {
await new Promise((r) => setTimeout(r, 5000));
const poll = await axios.get(this.resultUrl, {
params: { key: this.apiKey, action: "get", id: captchaId, json: 1 },
});
if (poll.data.status === 1) {
return { provider: this.name, success: true, solution: poll.data.request, elapsed: (Date.now() - start) / 1000 };
}
if (poll.data.request !== "CAPCHA_NOT_READY") {
return { provider: this.name, success: false, error: poll.data.request, elapsed: (Date.now() - start) / 1000 };
}
}
return { provider: this.name, success: false, error: "TIMEOUT", elapsed: (Date.now() - start) / 1000 };
} catch (err) {
return { provider: this.name, success: false, error: err.message, elapsed: (Date.now() - start) / 1000 };
}
}
}
async function parallelTest(current, captchaai, sitekey, pageurl, runs = 20) {
const results = { current: [], captchaai: [] };
for (let i = 0; i < runs; i++) {
const [currentResult, captchaaiResult] = await Promise.all([
current.solveRecaptcha(sitekey, pageurl),
captchaai.solveRecaptcha(sitekey, pageurl),
]);
results.current.push(currentResult);
results.captchaai.push(captchaaiResult);
console.log(`Run ${i + 1}/${runs}: ${current.name}=${currentResult.success ? "OK" : "FAIL"} ` +
`(${currentResult.elapsed.toFixed(1)}s) | ${captchaai.name}=${captchaaiResult.success ? "OK" : "FAIL"} ` +
`(${captchaaiResult.elapsed.toFixed(1)}s)`);
}
for (const [label, data] of Object.entries(results)) {
const successes = data.filter((r) => r.success).length;
const times = data.filter((r) => r.success).map((r) => r.elapsed);
const avgTime = times.length ? times.reduce((a, b) => a + b, 0) / times.length : 0;
console.log(`\n${label}: ${successes}/${runs} success (${((successes / runs) * 100).toFixed(1)}%), avg ${avgTime.toFixed(1)}s`);
}
}
// Run
const currentProvider = new CaptchaProvider("CurrentProvider", "https://current-provider.com/in.php", "https://current-provider.com/res.php", "current_key");
const captchaai = new CaptchaProvider("CaptchaAI", "https://ocr.captchaai.com/in.php", "https://ocr.captchaai.com/res.php", process.env.CAPTCHAAI_API_KEY);
parallelTest(currentProvider, captchaai, "SITE_KEY", "https://example.com", 20);
推荐的测试计划
| 阶段 | 期间 | 流量分割 | 目标 |
|---|---|---|---|
| 1. 验证 | 1天 | 0% 实时,仅限并行 | 验证 API 兼容性 |
| 2.阴影测试 | 3天 | 5% 至 CaptchaAI(有回退) | 收集基线指标 |
| 3. 加速 | 1周 | 25%→50%→75% | 各级监控 |
| 4、全割接 | —— | 100% CaptchaAI | 退役旧提供商 |
比较指标
| 公制 | 如何测量 |
|---|---|
| 成功率 | successful_solves / total_attempts × 100 |
| 平均求解时间 | 从提交到收到解决方案的时间 |
| P95 求解时间 | 求解时间的第 95 个百分位 |
| 按类型划分的错误率 | 分别统计每个错误代码 |
| 每次解决的成本 | 总支出/成功解决问题 |
| 令牌有效性 | 返回的令牌实际上在目标站点上有效吗? |
故障排除
| 问题 | 原因 | 处理方式 |
|---|---|---|
| CaptchaAI 并行测试较慢 | 网络延迟差异 | 从生产服务器而不是本地计算机进行测试 |
| 成功率不同 | 提供商求解器池各不相同 | 运行 50 多个解决方案以获得统计显着性 |
| 解决方案适用于一个提供商,但不适用于其他提供商 | 令牌过期时间 | 收到令牌后立即使用 |
| 并行测试使成本加倍 | 解决相同的验证码两次 | 测试期预算;它比失败的迁移便宜 |
常问问题
我需要运行多少次测试才能获得可靠的数据?
每个提供商至少 50 个解决方案。要获得具有统计显着性的结果,请在一天中的不同时间运行 100 多个求解以考虑方差。
我应该使用代理还是无代理进行测试?
如果您在生产中同时使用两者,请测试两者。代理质量对不同提供商的解决率的影响不同 - 测试您的实际代理配置。
如果 CaptchaAI 在一种验证码类型上表现较差怎么办?
运行特定于类型的测试。提供商可能擅长 reCAPTCHA,但在 hCaptcha 方面落后。如果您使用多种类型,则根据您的实际流量分布计算权重结果。
相关文章
下一步
开始无风险的并行测试 -创建您的 CaptchaAI 帐户并将性能与实际数据进行比较。
相关指南: