Tutorials

限速并发：CAPTCHA API 调用的令牌桶

自动化 Python 所有 CAPTCHA 类型 性能
作者 CaptchaAI 中文团队
发布时间 Jul 24, 2026
限速并发：CAPTCHA API 调用的令牌桶
特色图片不可用

并发一开大，日志里就刷 ERROR_TOO_MUCH_REQUESTS；把线程池从 30 调到 10，吞吐掉一半，错误照样冒。

该控的不是线程数，而是每秒往 in.php 提交多少个新任务。

令牌桶就干这件事：按固定速率补充令牌，提交前先取一个，取不到就等。

你最终拿到一个能写进配置的确定值：每秒最多 20 次。

限流算法怎么选

算法 行为 适合场景
令牌桶 速率平稳，允许突发 验证码 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
概念 含义
容量 允许的最大突发量
补充率 每秒稳定放行的请求数
排队 桶空时请求等待，不拒绝，只节流

注意：这里的“令牌”是限流器的计数单位，和识别返回的 token 无关。

Python 实现

线程安全的令牌桶

用 time.monotonic() 而非 time.time()，避免系统校时算出负的时间差。

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}")

Node.js 实现

异步令牌桶

单线程不需要锁，await 等到令牌可用即可。

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/秒
常规自动化 20 10/秒
大批量流水线 50 30/秒
满负荷吞吐 100 50/秒

容量取补充率的 2 倍，相当于允许 2 秒突发。

从保守值起步，盯着错误率往上加。

场景：跨境比价采集

海外站点多是 reCAPTCHA v2 和 Cloudflare Turnstile，国内站点则以 GeeTest（极验）v3 为主。

假设团队用 ADVANCE 套餐（$90/月，50 线程），这里就有两道闸门：线程数管并发上限，令牌桶管每秒新进来多少个。

补充率的起点直接算：线程数 ÷ 单任务实测耗时。50 线程、实测 25 秒，稳态就是每秒 2 个，再往上加只变成排队延迟。

采集前确认数据在授权范围内：网络安全法与 robots 协议都适用。

限流排查清单

现象 原因 处理方式
加了限流仍被限速 补充率高于接口允许值 调低补充率
延迟明显变高 令牌取空，都在等补充 调大容量
内存持续增长 等待队列越堆越长 队列设上限，超出即拒
多进程各限各的 令牌桶只在单进程内存里 换 Redis 版分布式令牌桶

常见问题

令牌桶容量和 CaptchaAI 的线程数是一回事吗？

不是。线程数是并发上限，由套餐决定：BASIC $15/月 5 线程，ADVANCE $90/月 50 线程，线程内不限次数。

令牌桶管的是每秒提交量。两者不匹配时，多出来的提交只会变成延迟。

补充率一开始设多少合适？

先按“线程数 ÷ 单任务实测耗时”算理论值，打七折起步。

跑半小时看错误率，之后每次上调 20% 左右。

提交和轮询都要限流吗？

只限提交。

轮询很轻，循环里的 time.sleep(5) 已经限住频率，再加一层只会拖长返回时间。

相关文章

下一步

获取 CaptchaAI API Key，先让令牌桶的补充率和套餐线程数对上号。

相关指南：

该文章已禁用评论。