CAPTCHA API 断路器模式实现指南

调用 reCAPTCHA、Turnstile 这类验证码 API 时，接口偶尔变慢或连续报错，代码还在不停重试，额度白白消耗，用户端等待也跟着变长——这是很多团队都遇到过的问题。

断路器模式（Circuit Breaker，也叫熔断机制）：连续失败超过阈值后先暂停调用，给下游一点恢复时间，再用一次试探请求判断能不能恢复，避免级联故障拖垮整条链路。

reCAPTCHA 依赖 Google 托管资源，出海团队从国内网络访问境外接口本身就不太稳定，断路器几乎是这类链路的标配组件。

断路器的三种状态：关闭、打开、半开

断路器本质上是一个简单的状态机，一共三种状态：

状态	触发条件	行为
关闭（Closed）	默认状态	请求照常发出，失败计数
打开（Open）	失败达阈值	请求直接拒绝，不再打到 API
半开（Half-Open）	冷却已过	放行一次试探；成功回关闭，失败回打开

Python 实现：给验证码请求加上熔断保护

下面的 CircuitBreaker 类用线程锁保护状态切换，可直接套在现有请求函数外层：

import time
import threading
import requests

SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"
API_KEY = "YOUR_API_KEY"


class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = 0
        self.state = "closed"  # closed, open, half-open
        self._lock = threading.Lock()

    def call(self, func, *args, **kwargs):
        with self._lock:
            if self.state == "open":
                if time.time() - self.last_failure_time > self.recovery_timeout:
                    self.state = "half-open"
                    print("[circuit] State: half-open — testing one request")
                else:
                    remaining = self.recovery_timeout - (
                        time.time() - self.last_failure_time
                    )
                    raise CircuitOpenError(
                        f"Circuit open — retry in {remaining:.0f}s"
                    )

        try:
            result = func(*args, **kwargs)
            with self._lock:
                self.failure_count = 0
                if self.state == "half-open":
                    print("[circuit] State: closed — API recovered")
                self.state = "closed"
            return result
        except Exception as e:
            with self._lock:
                self.failure_count += 1
                self.last_failure_time = time.time()
                if self.failure_count >= self.failure_threshold:
                    self.state = "open"
                    print(
                        f"[circuit] State: open — "
                        f"{self.failure_count} failures"
                    )
            raise


class CircuitOpenError(Exception):
    pass


def solve_captcha(sitekey, page_url):
    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:
        raise Exception(f"Submit error: {data['request']}")

    task_id = data["request"]
    for _ in range(24):
        time.sleep(5)
        poll = requests.get(RESULT_URL, params={
            "key": API_KEY,
            "action": "get",
            "id": task_id,
            "json": "1",
        }, timeout=15).json()
        if poll["status"] == 1:
            return poll["request"]
        if poll["request"] != "CAPCHA_NOT_READY":
            raise Exception(f"Poll error: {poll['request']}")
    raise TimeoutError(f"Task {task_id} timed out")


# Usage
breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30)

for i in range(10):
    try:
        token = breaker.call(
            solve_captcha, "6Le-SITEKEY", "https://example.com"
        )
        print(f"[task-{i}] Solved: {token[:40]}...")
    except CircuitOpenError as e:
        print(f"[task-{i}] Skipped: {e}")
    except Exception as e:
        print(f"[task-{i}] Failed: {e}")

预期输出：

[task-0] Solved: 03AGdBq26ZfPxL...
[task-1] Solved: 03AGdBq27AbCdE...
[task-2] Failed: Submit error: ERROR_NO_SLOT_AVAILABLE
[task-3] Failed: Submit error: ERROR_NO_SLOT_AVAILABLE
[task-4] Failed: Submit error: ERROR_NO_SLOT_AVAILABLE
[circuit] State: open — 3 failures
[task-5] Skipped: Circuit open — retry in 28s
[task-6] Skipped: Circuit open — retry in 25s
...
[circuit] State: half-open — testing one request
[task-8] Solved: 03AGdBq28FgHiJ...
[circuit] State: closed — API recovered

JavaScript 实现

Node.js 服务端思路一样，只是把线程锁换成了单线程事件循环的顺序执行：

class CircuitBreaker {
  constructor(options = {}) {
    this.failureThreshold = options.failureThreshold || 5;
    this.recoveryTimeout = options.recoveryTimeout || 60000;
    this.failureCount = 0;
    this.lastFailureTime = 0;
    this.state = 'closed';
  }

  async call(fn, ...args) {
    if (this.state === 'open') {
      if (Date.now() - this.lastFailureTime > this.recoveryTimeout) {
        this.state = 'half-open';
        console.log('[circuit] State: half-open');
      } else {
        const remaining = this.recoveryTimeout - (Date.now() - this.lastFailureTime);
        throw new Error(`Circuit open — retry in ${Math.ceil(remaining / 1000)}s`);
      }
    }

    try {
      const result = await fn(...args);
      this.failureCount = 0;
      if (this.state === 'half-open') {
        console.log('[circuit] State: closed — recovered');
      }
      this.state = 'closed';
      return result;
    } catch (error) {
      this.failureCount++;
      this.lastFailureTime = Date.now();
      if (this.failureCount >= this.failureThreshold) {
        this.state = 'open';
        console.log(`[circuit] State: open — ${this.failureCount} failures`);
      }
      throw error;
    }
  }
}

// Usage
const axios = require('axios');

const API_KEY = 'YOUR_API_KEY';
const breaker = new CircuitBreaker({ failureThreshold: 3, recoveryTimeout: 30000 });

async function solveCaptcha(sitekey, pageurl) {
  const submit = await axios.post('https://ocr.captchaai.com/in.php', null, {
    params: { key: API_KEY, method: 'userrecaptcha', googlekey: sitekey, pageurl, json: 1 }
  });

  if (submit.data.status !== 1) throw new Error(submit.data.request);
  const taskId = submit.data.request;

  for (let i = 0; i < 24; i++) {
    await new Promise(r => setTimeout(r, 5000));
    const poll = await axios.get('https://ocr.captchaai.com/res.php', {
      params: { key: API_KEY, action: 'get', id: taskId, json: 1 }
    });
    if (poll.data.status === 1) return poll.data.request;
    if (poll.data.request !== 'CAPCHA_NOT_READY') throw new Error(poll.data.request);
  }
  throw new Error('Timeout');
}

(async () => {
  for (let i = 0; i < 10; i++) {
    try {
      const token = await breaker.call(solveCaptcha, '6Le-SITEKEY', 'https://example.com');
      console.log(`[task-${i}] Solved: ${token.substring(0, 40)}...`);
    } catch (err) {
      console.log(`[task-${i}] ${err.message}`);
    }
  }
})();

一个真实场景：跨境电商结账链路

某跨境电商结账页脚本每天触发数千次 reCAPTCHA 验证，走境外线路访问 Google 托管资源，高峰期连续超时会让重试和等待一起失控：

连续 3 次失败即打开电路，暂停 30 秒；
冷却期间把任务放入队列，不再硬打 API；
半开状态试探一次判断是否恢复。

只需在 solve_captcha 外层套一层 CircuitBreaker，业务逻辑不用动。

断路器阈值怎么选：低流量 vs 高流量

参数	低流量（< 10 次/分钟）	高流量（> 100 次/分钟）
`failure_threshold`	3	10
`recovery_timeout`	30 秒	60 秒

阈值太低会误判正常抖动为故障，太高又会让电路在 API 真正故障时继续硬撞。经验做法：低流量用更低阈值（3 次）快速止损，高流量放宽到 10 次左右。

和重试逻辑搭配使用

重试逻辑应该放在断路器内部，让断路器只统计重试全部失败后的最终失败次数：

def solve_with_retry(sitekey, page_url, max_retries=2):
    for attempt in range(max_retries + 1):
        try:
            return solve_captcha(sitekey, page_url)
        except Exception:
            if attempt == max_retries:
                raise
            time.sleep(2 ** attempt)

# Circuit breaker wraps the retry function
token = breaker.call(solve_with_retry, "6Le-SITEKEY", "https://example.com")

短暂的单次超时会被重试悄悄吸收，只有连续多次真正失败才会触发熔断。

常见故障排查

问题	原因	处理方式
电路跳闸太快	阈值设置过低	调高 `failure_threshold`
电路一直恢复不了	`recovery_timeout` 设置过长	缩短到 30–60 秒
多线程环境下出现竞态	状态没有加锁	Python 用 `threading.Lock`，其他语言用原子操作
部分故障期间所有请求都被拦截	提交和轮询共用同一个断路器	给提交接口和轮询接口分别配置独立的断路器

常见问题

提交和轮询要不要用两个独立的断路器？

大规模系统建议分开。提交接口出问题时轮询接口可能仍正常，独立断路器能做更细粒度的控制。

电路处于“打开”状态时，验证码任务该怎么处理？

常见做法是排队稍后重试、展示降级 UI，或直接跳过。参考解决失败时的优雅降级。

断路器和普通的重试机制有什么区别？

重试只是失败了再试一次，不记忆历史失败次数；断路器持续统计失败次数，超过阈值就直接停止调用，两者是配合关系。

半开状态下的试探请求又失败了会怎样？

电路立刻回到打开状态，重新计时冷却周期，避免反复试探打断恢复过程。

多进程或多实例部署时，断路器状态需要共享吗？

各实例独立处理队列时，本地状态通常够用；多个实例共同面对同一下游、流量又大时，放进 Redis 等共享存储会更准确。

用 CaptchaAI 构建更稳定的验证码调用链路

把断路器包在现有请求函数外层，几行代码就能让整条流程在 API 抖动时更稳。前往 CaptchaAI 官网获取你的 API Key。

CAPTCHA API 调用的断路器模式

断路器的三种状态：关闭、打开、半开

Python 实现：给验证码请求加上熔断保护

JavaScript 实现

一个真实场景：跨境电商结账链路

断路器阈值怎么选：低流量 vs 高流量

和重试逻辑搭配使用

常见故障排查

常见问题

提交和轮询要不要用两个独立的断路器？

电路处于“打开”状态时，验证码任务该怎么处理？

断路器和普通的重试机制有什么区别？

半开状态下的试探请求又失败了会怎样？

多进程或多实例部署时，断路器状态需要共享吗？

用 CaptchaAI 构建更稳定的验证码调用链路

相关阅读

使用 CaptchaAI 构建客户端验证码管道

弹出模式中的验证码：检测和token 提交

高需求 e-commerce checkout 的 CAPTCHA 测试

使用 CaptchaAI 进行 QA 浏览器配置文件隔离

浏览器自动化验证码失败但 API 有效：调试指南

使用 CaptchaAI 构建自动化测试管道

断路器的三种状态：关闭、打开、半开

Python 实现：给验证码请求加上熔断保护

JavaScript 实现

一个真实场景：跨境电商结账链路

断路器阈值怎么选：低流量 vs 高流量

和重试逻辑搭配使用

常见故障排查

常见问题

提交和轮询要不要用两个独立的断路器？

电路处于“打开”状态时，验证码任务该怎么处理？

断路器和普通的重试机制有什么区别？

半开状态下的试探请求又失败了会怎样？

多进程或多实例部署时，断路器状态需要共享吗？

用 CaptchaAI 构建更稳定的验证码调用链路

相关阅读

相关文章

使用 CaptchaAI 构建客户端验证码管道

弹出模式中的验证码：检测和token 提交

高需求 e-commerce checkout 的 CAPTCHA 测试

使用 CaptchaAI 进行 QA 浏览器配置文件隔离

浏览器自动化验证码失败但 API 有效：调试指南

使用 CaptchaAI 构建自动化测试管道