不受控制的并发尽可能快地发送请求。这会导致 ERROR_TOO_MUCH_REQUESTS、API 余额浪费和不可预测的成本。令牌桶可让您设置精确的速率——“每秒提交不超过 20 次”——同时在容量可用时仍允许短时间突发。
令牌桶的工作原理
[Bucket] capacity=20, refill=10/sec
Time 0: ████████████████████ 20 tokens available
→ 15 requests consume 15 tokens
Time 0: █████ 5 tokens remain
Time 1s: ███████████████ 15 tokens (5 + 10 refilled)
→ 15 requests consume 15 tokens
Time 1s: (empty) 0 tokens
Time 2s: ██████████ 10 tokens (0 + 10 refilled)
→ Request waits if bucket is empty
主要特性：
- 容量 – 最大突发大小
- 重新填充率 – 每秒持续请求数
- 请求在桶为空时等待（没有拒绝，只是节流）
Python实现
线程安全令牌桶
import time
import threading
class TokenBucket:
def __init__(self, capacity, refill_rate):
"""
Args:
capacity: Maximum tokens (burst size)
refill_rate: Tokens added per second
"""
self.capacity = capacity
self.refill_rate = refill_rate
self.tokens = capacity
self.last_refill = time.monotonic()
self.lock = threading.Lock()
def acquire(self, timeout=None):
"""Block until a token is available."""
deadline = time.monotonic() + timeout if timeout else float("inf")
while True:
with self.lock:
self._refill()
if self.tokens >= 1:
self.tokens -= 1
return True
# Check timeout
if time.monotonic() >= deadline:
return False
# Wait before retrying (avoid busy loop)
time.sleep(min(1.0 / self.refill_rate, 0.1))
def _refill(self):
now = time.monotonic()
elapsed = now - self.last_refill
new_tokens = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill = now
速率受限的验证码求解器
import os
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
API_KEY = os.environ["CAPTCHAAI_API_KEY"]
# Allow 10 submissions/sec with burst of 20
rate_limiter = TokenBucket(capacity=20, refill_rate=10)
def solve_captcha_rate_limited(sitekey, pageurl):
"""Solve with rate limiting on submission."""
# Wait for token before submitting
rate_limiter.acquire()
resp = requests.post("https://ocr.captchaai.com/in.php", data={
"key": API_KEY,
"method": "userrecaptcha",
"googlekey": sitekey,
"pageurl": pageurl,
"json": 1
})
data = resp.json()
if data.get("status") != 1:
raise RuntimeError(data.get("request"))
captcha_id = data["request"]
# Polling doesn't need rate limiting (separate concern)
for _ in range(60):
time.sleep(5)
result = requests.get("https://ocr.captchaai.com/res.php", params={
"key": API_KEY, "action": "get", "id": captcha_id, "json": 1
}).json()
if result.get("status") == 1:
return result["request"]
if result.get("request") != "CAPCHA_NOT_READY":
raise RuntimeError(result.get("request"))
raise TimeoutError("Solve timeout")
# Run 100 tasks through rate limiter
tasks = [
{"sitekey": "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
"pageurl": f"https://example.com/p/{i}"}
for i in range(100)
]
with ThreadPoolExecutor(max_workers=30) as executor:
futures = {
executor.submit(
solve_captcha_rate_limited, t["sitekey"], t["pageurl"]
): t for t in tasks
}
for future in as_completed(futures):
task = futures[future]
try:
solution = future.result()
print(f"[OK] {task['pageurl']}")
except Exception as e:
print(f"[ERR] {task['pageurl']}: {e}")
JavaScript 实现
异步令牌桶
class TokenBucket {
constructor(capacity, refillRate) {
this.capacity = capacity;
this.refillRate = refillRate; // tokens per second
this.tokens = capacity;
this.lastRefill = Date.now();
this.waitQueue = [];
}
_refill() {
const now = Date.now();
const elapsed = (now - this.lastRefill) / 1000;
this.tokens = Math.min(this.capacity, this.tokens + elapsed * this.refillRate);
this.lastRefill = now;
}
async acquire() {
this._refill();
if (this.tokens >= 1) {
this.tokens -= 1;
return;
}
// Wait until a token is available
const waitTime = ((1 - this.tokens) / this.refillRate) * 1000;
await new Promise((resolve) => setTimeout(resolve, waitTime));
this._refill();
this.tokens -= 1;
}
}
限速批量求解器
const axios = require("axios");
const API_KEY = process.env.CAPTCHAAI_API_KEY;
const rateLimiter = new TokenBucket(20, 10); // 20 burst, 10/sec sustained
function sleep(ms) {
return new Promise((resolve) => setTimeout(resolve, ms));
}
async function solveCaptchaLimited(sitekey, pageurl) {
// Wait for rate limit token
await rateLimiter.acquire();
const submitResp = await axios.post(
"https://ocr.captchaai.com/in.php",
null,
{
params: {
key: API_KEY,
method: "userrecaptcha",
googlekey: sitekey,
pageurl: pageurl,
json: 1,
},
}
);
if (submitResp.data.status !== 1) {
throw new Error(submitResp.data.request);
}
const captchaId = submitResp.data.request;
for (let i = 0; i < 60; i++) {
await sleep(5000);
const result = await axios.get("https://ocr.captchaai.com/res.php", {
params: { key: API_KEY, action: "get", id: captchaId, json: 1 },
});
if (result.data.status === 1) return result.data.request;
if (result.data.request !== "CAPCHA_NOT_READY") {
throw new Error(result.data.request);
}
}
throw new Error("TIMEOUT");
}
// Solve 100 tasks — rate limiter ensures max 10 submissions/sec
async function batchSolve(tasks) {
const results = await Promise.allSettled(
tasks.map((t) => solveCaptchaLimited(t.sitekey, t.pageurl))
);
const solved = results.filter((r) => r.status === "fulfilled").length;
const failed = results.filter((r) => r.status === "rejected").length;
console.log(`Solved: ${solved}, Failed: ${failed}`);
}
选择参数
| 工作量 | 容量（突发） | 补充率（持续） |
|---|---|---|
| 轻刮 | 5 | 2/sec |
| 标准自动化 | 20 | 10/sec |
| 大容量管道 | 50 | 30/sec |
| 最大吞吐量 | 100 | 50/sec |
经验法则：
- 将容量设置为 2 × 填充率（允许 2 秒突发）
- 开始保守，在监控错误率的同时增加
- 仅对提交进行速率限制 - 轮询是轻量级且自我限制的
令牌桶与其他算法
| 算法 | 行为 | 最适合 |
|---|---|---|
| 令牌桶 | 具有突发津贴的平稳速率 | 验证码 API 调用 |
| 漏水桶 | 固定输出速率，无突发 | 严格的费率要求 |
| 固定窗 | 每个时间窗口计数，边缘突发 | 简单计数器 |
| 推拉窗 | 滚动期间计数 | 准确的费率执行 |
令牌桶是最好的默认设置 - 它允许自然爆发（抓取工具一次找到 20 个验证码），同时强制执行持续的速率。
故障排除
| 问题 | 原因 | 处理方式 |
|---|---|---|
| 请求仍然受到限制 | 速率限制器设置高于 API 允许的值 | 较低的补充率以符合 CaptchaAI 的限制 |
| 请求延迟高 | 代币已用完，等待补充 | 增加突发场景的容量 |
| 记忆力增长 | 等待队列累积 | 设置最大队列大小；拒绝多余的请求 |
| 速率限制器未跨进程共享 | 仅限内存中 | 使用基于Redis的令牌桶进行分布式限速 |
常问问题
我应该限制提交、投票或两者的速率吗？
仅限速提交。轮询请求是轻量级的，并且通过 time.sleep(5) 进行自我调节。过度限制轮询会增加解决延迟，但没有任何好处。
尽管有速率限制，我如何处理 ERROR_TOO_MUCH_REQUESTS？
您的速率限制设置得太高。降低补充率。还要检查多个进程是否共享相同的 API 密钥——所有进程的聚合速率。
我可以对每个验证码类型使用速率限制器吗？
是的 - 为不同的验证码类型创建单独的令牌桶。这可以防止大量 reCAPTCHA v2 任务导致 Turnstile 提交不足。
相关文章
下一步
构建速率受控的验证码解决方案 -获取您的 CaptchaAI API 密钥并实施可持续的请求率。
相关指南：
