验证码识别量上来后,很多团队先加线程、升级套餐。但真正卡住流水线的,常常是只用了一个 API 密钥——限流、清零或禁用,任务就停摆。把流量分散到多个密钥,让轮换策略接管故障转移更稳妥。
比如 618、双11 期间,只用一个 STANDARD($30/月,15 线程)账号很容易同时打满余额和限流;把 3 个账号的密钥做加权轮换,通常就够扛住突发流量,不必升级到 ADVANCE($90/月,50 线程)。
三种轮换策略,复杂度递增:
- 轮询轮换:顺序循环,最简单。
- 按余额加权轮换:余额高的账号多扛流量。
- 故障转移轮换:出错即切下一个,常与前两种搭配。
轮询轮换:最简单的负载分摊
最简单的策略是依次使用每个密钥:
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)
_lock不能省略:多线程同时读写会出现竞态条件。
故障转移轮换:一个密钥失败就切下一个
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")
达到
max_attempts仍失败,说明密钥已耗尽,需要人工介入。
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');
}
Python 和 JavaScript 两个版本的重试逻辑完全对应:网络异常和业务失败都会触发换密钥重试,达到 max_attempts 才抛出异常。
用环境变量管理密钥,别硬编码
- API 密钥不能写进代码仓库,一律从环境变量读取。
- 只放进
.env或 CI/CD 密钥库(如 GitHub Actions Secrets),加入.gitignore。 - 怀疑泄露时立即停用该密钥、生成新密钥,并从轮换池中移除旧密钥。
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);
最短路径:读取环境变量 → 拆分成列表 → 传给
KeyRotator。
定时刷新余额,别等到密钥用尽才发现
长期运行的服务应定期主动刷新余额:
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
5 分钟是经验值:密钥池小可缩短到 1–2 分钟,密钥多时适当拉长,避免频繁调用 getbalance。
三种策略怎么选
按场景对号入座:
| 场景 | 推荐策略 | 密钥数量参考 |
|---|---|---|
| 单账号 STANDARD($30/月,15 线程),流量平稳 | 轮询轮换 | 2–3 个 |
| 多账号套餐,余额差异大 | 按余额加权轮换 | 3–5 个 |
| 618、双11 大促流量突增 | 加权轮换 + 故障转移兜底 | 3 个以上 |
故障排查
先看余额,再看索引,再核对错误码。
| 问题 | 原因 | 处理方式 |
|---|---|---|
| 所有密钥都被禁用 | 所有账户余额都是零 | 给账户充值,检查是否命中 ERROR_ZERO_BALANCE |
| 一直在用同一个密钥 | 轮询索引没有正确推进 | 确认索引更新有加锁保护,排查线程安全问题 |
| 密钥被错误禁用 | 把临时性错误当成了永久性错误 | 只在 ERROR_WRONG_USER_KEY、ERROR_ZERO_BALANCE、ERROR_IP_NOT_ALLOWED 上禁用密钥 |
常见问题
落地时最容易卡壳的几个问题:
应该用几个 API 密钥做轮换?
两个密钥提供基本故障转移,三个以上可做负载分摊。识别量 1000+ 次/天时,建议 3–5 个密钥配合加权。
密钥轮换会增加我的账单吗?
不会。CaptchaAI 按线程数计费,不按密钥数计费。同账号轮换无额外费用;跨账号轮换,各账号按自身套餐计费。
怎么避免轮换一直卡在一个已失效的密钥上?
只对 ERROR_WRONG_USER_KEY、ERROR_ZERO_BALANCE 等永久性错误禁用密钥,网络超时等临时错误不应永久踢出。
可以把不同 CaptchaAI 账号的密钥放进同一个轮换池吗?
可以,每个密钥的余额和限流额度相互独立,轮换器会分别处理。
密钥被禁用后要不要设置告警?
小规模场景,日志够用;密钥多或对故障敏感,建议把禁用事件推到告警渠道(如 Slack webhook)。
用多密钥轮换稳定你的验证码识别流水线
机制可按流量规模逐步上:
- 先用 2–3 个密钥轮询。
- 流量涨了切到加权。
- 对故障敏感再叠加告警。
注册 CaptchaAI,拿到密钥即可接入上面的轮换代码:CaptchaAI 官网。
相关指南
延伸阅读:
- CaptchaAI IP 白名单和 API 密钥安全
- 保护环境变量中的 CaptchaAI 凭证
- CaptchaAI 余额检查和自动充值