通过单个 API 密钥运行所有流量会产生单点故障。如果密钥失去平衡、达到速率限制或被停用，则整个管道都会停止。密钥轮换将负载分布到多个密钥并提供自动故障转移。
循环赛轮换
最简单的策略——均匀地循环按键：
Python
import itertools
import requests
API_KEYS = [
"KEY_ACCOUNT_1",
"KEY_ACCOUNT_2",
"KEY_ACCOUNT_3",
]
key_cycle = itertools.cycle(API_KEYS)
def get_next_key():
return next(key_cycle)
def solve_captcha(sitekey, page_url):
api_key = get_next_key()
resp = requests.post("https://ocr.captchaai.com/in.php", data={
"key": api_key,
"method": "userrecaptcha",
"googlekey": sitekey,
"pageurl": page_url,
"json": "1",
})
data = resp.json()
if data["status"] != 1:
raise Exception(f"[{api_key[:8]}...] {data['request']}")
print(f"Submitted with key {api_key[:8]}...")
return data["request"], api_key
task_id, used_key = solve_captcha("6Le-SITEKEY", "https://example.com")
具有平衡意识的加权旋转
将更多流量路由到余额较高的密钥：
import random
import requests
import threading
SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"
class KeyRotator:
def __init__(self, keys):
self.keys = {k: {"balance": 0, "failures": 0, "disabled": False} for k in keys}
self._lock = threading.Lock()
self.refresh_balances()
def refresh_balances(self):
for key in self.keys:
try:
resp = requests.get(RESULT_URL, params={
"key": key, "action": "getbalance", "json": "1"
}, timeout=10).json()
if resp["status"] == 1:
self.keys[key]["balance"] = float(resp["request"])
self.keys[key]["disabled"] = False
else:
self.keys[key]["disabled"] = True
except Exception:
self.keys[key]["disabled"] = True
def get_key(self):
with self._lock:
available = {
k: v for k, v in self.keys.items()
if not v["disabled"] and v["balance"] > 0.01
}
if not available:
raise Exception("No API keys with balance available")
# Weighted random by balance
keys = list(available.keys())
weights = [available[k]["balance"] for k in keys]
return random.choices(keys, weights=weights, k=1)[0]
def report_failure(self, key, error_code):
with self._lock:
self.keys[key]["failures"] += 1
if error_code in ("ERROR_WRONG_USER_KEY", "ERROR_KEY_DOES_NOT_EXIST",
"ERROR_ZERO_BALANCE", "ERROR_IP_NOT_ALLOWED"):
self.keys[key]["disabled"] = True
print(f"[rotator] Disabled key {key[:8]}...: {error_code}")
def report_success(self, key, cost=0.003):
with self._lock:
self.keys[key]["balance"] -= cost
self.keys[key]["failures"] = 0
rotator = KeyRotator(["KEY_1", "KEY_2", "KEY_3"])
# Usage
api_key = rotator.get_key()
# ... solve captcha ...
rotator.report_success(api_key)
故障转移轮换
当一个键失败时尝试下一个键：
Python
def solve_with_failover(sitekey, page_url, max_attempts=3):
for attempt in range(max_attempts):
api_key = rotator.get_key()
try:
resp = requests.post(SUBMIT_URL, data={
"key": api_key,
"method": "userrecaptcha",
"googlekey": sitekey,
"pageurl": page_url,
"json": "1",
}, timeout=15)
data = resp.json()
if data["status"] != 1:
rotator.report_failure(api_key, data["request"])
continue
rotator.report_success(api_key)
return data["request"], api_key
except requests.RequestException:
rotator.report_failure(api_key, "NETWORK_ERROR")
continue
raise Exception(f"All {max_attempts} keys failed")
JavaScript
const axios = require('axios');
class KeyRotator {
constructor(keys) {
this.keys = keys.map(k => ({ key: k, disabled: false, failures: 0 }));
this.index = 0;
}
getKey() {
const available = this.keys.filter(k => !k.disabled);
if (available.length === 0) throw new Error('No API keys available');
const entry = available[this.index % available.length];
this.index++;
return entry.key;
}
disable(key, reason) {
const entry = this.keys.find(k => k.key === key);
if (entry) {
entry.disabled = true;
console.log(`[rotator] Disabled ${key.substring(0, 8)}...: ${reason}`);
}
}
}
const rotator = new KeyRotator(['KEY_1', 'KEY_2', 'KEY_3']);
async function solveWithFailover(sitekey, pageurl, maxAttempts = 3) {
for (let i = 0; i < maxAttempts; i++) {
const apiKey = rotator.getKey();
try {
const resp = await axios.post('https://ocr.captchaai.com/in.php', null, {
params: { key: apiKey, method: 'userrecaptcha', googlekey: sitekey, pageurl, json: 1 }
});
if (resp.data.status !== 1) {
rotator.disable(apiKey, resp.data.request);
continue;
}
return { taskId: resp.data.request, apiKey };
} catch (err) {
rotator.disable(apiKey, 'NETWORK_ERROR');
}
}
throw new Error('All keys failed');
}
从环境变量加载密钥
切勿对 API 密钥进行硬编码。从环境加载：
import os
API_KEYS = os.environ["CAPTCHAAI_KEYS"].split(",")
# Set: CAPTCHAAI_KEYS=key1,key2,key3
rotator = KeyRotator(API_KEYS)
const API_KEYS = process.env.CAPTCHAAI_KEYS.split(',');
const rotator = new KeyRotator(API_KEYS);
预定余额刷新
对于长时间运行的进程，定期刷新平衡：
import threading
def periodic_refresh(rotator, interval=300):
def refresh():
while True:
rotator.refresh_balances()
for key, info in rotator.keys.items():
print(f" {key[:8]}...: ${info['balance']:.2f} "
f"{'(disabled)' if info['disabled'] else '(active)'}")
threading.Event().wait(interval)
t = threading.Thread(target=refresh, daemon=True)
t.start()
periodic_refresh(rotator, interval=300) # every 5 minutes
故障排除
| 问题 | 原因 | 处理方式 |
|---|---|---|
| 所有按键均已禁用 | 所有账户余额为零 | 充值账户，查看ERROR_ZERO_BALANCE |
| 始终使用相同的密钥 | 循环索引未推进 | 用锁检查螺纹安全 |
| 密钥被错误禁用 | 临时错误被视为永久错误 | 仅在 ERROR_WRONG_USER_KEY、ERROR_ZERO_BALANCE、ERROR_IP_NOT_ALLOWED 上禁用 |
常问问题
我应该使用多少个 API 密钥？
两个键提供基本的故障转移。三个或更多键允许负载分配。对于大批量操作（1000+ 解决 /day），请考虑 3-5 个密钥。
我可以使用不同 CaptchaAI 账户的密钥吗？
是的。每个密钥都有自己的余额和速率限制。旋转者独立对待它们。
通过多键轮换扩展您的验证码解决方案
获取您的 API 密钥：验证码网站。
相关指南
- CaptchaAI IP 白名单和 API 密钥安全
- 保护环境变量中的 CaptchaAI 凭证
- CaptchaAI 余额检查和自动充值
