使用 CaptchaAI 的多区域验证码解决架构

Worker 只部署在一个区域，却要给全球目标站点解决验证码，迟早撞上延迟高、单点故障两个问题。多区域架构把 Worker 拆到离目标站点更近的地方，用同一个 API Key 统一调度。

先判断要不要上多区域

不是所有项目都需要这套复杂度，先对照下表：

举个例子：团队总部在杭州、业务面向北美电商站点，Worker 若留在国内机房，延迟高且 reCAPTCHA 依赖的 Google 脚本也未必稳定加载；迁到新加坡或美区节点，两个问题一起解决。

整体架构长什么样

                    [Task Router]
                    (Route53 / Load Balancer)
                    ↙        ↓        ↘
        [US-East]      [EU-West]     [AP-Southeast]
        Workers        Workers        Workers
            ↓              ↓              ↓
        [CaptchaAI API] ← shared API key
            ↓              ↓              ↓
        [Central DB / Queue]
        (Results aggregation)

每个区域跑独立的 Worker，共用同一个 API Key，结果统一推送到中央存储，区域之间互不牵连。

部署区域感知的 Worker

Python Worker（带区域标签）

import os
import time
import requests

API_KEY = os.environ["CAPTCHAAI_API_KEY"]
REGION = os.environ.get("WORKER_REGION", "us-east-1")
RESULT_QUEUE_URL = os.environ["RESULT_QUEUE_URL"]


def solve_captcha(task):
    """Solve CAPTCHA and tag with region metadata."""
    start = time.time()

    resp = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": task["method"],
        "googlekey": task["sitekey"],
        "pageurl": task["pageurl"],
        "json": 1
    })
    data = resp.json()

    if data.get("status") != 1:
        return {
            "task_id": task["task_id"],
            "error": data.get("request"),
            "region": REGION
        }

    captcha_id = data["request"]

    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 {
                "task_id": task["task_id"],
                "solution": result["request"],
                "region": REGION,
                "duration": time.time() - start,
                "api_latency_ms": round((time.time() - start) * 1000)
            }
        if result.get("request") != "CAPCHA_NOT_READY":
            return {
                "task_id": task["task_id"],
                "error": result.get("request"),
                "region": REGION
            }

    return {"task_id": task["task_id"], "error": "TIMEOUT", "region": REGION}

带上 region 字段，出问题时一眼看出是哪个区域的锅。

任务路由：把请求送到最近的区域

按目标站点域名后缀分流：

from urllib.parse import urlparse

# Region mapping by target site TLD/domain
REGION_MAP = {
    ".co.uk": "eu-west-1",
    ".de": "eu-central-1",
    ".fr": "eu-west-3",
    ".jp": "ap-northeast-1",
    ".com.au": "ap-southeast-2",
    ".com": "us-east-1",  # Default
}

REGION_QUEUES = {
    "us-east-1": "sqs://captcha-tasks-us-east",
    "eu-west-1": "sqs://captcha-tasks-eu-west",
    "ap-southeast-1": "sqs://captcha-tasks-ap-southeast",
}


def route_task(task):
    """Route task to the closest regional queue."""
    domain = urlparse(task["pageurl"]).netloc

    target_region = "us-east-1"  # Default
    for suffix, region in REGION_MAP.items():
        if domain.endswith(suffix):
            target_region = region
            break

    queue = REGION_QUEUES.get(target_region, REGION_QUEUES["us-east-1"])
    send_to_queue(queue, task)
    return target_region

基础设施：一次把三个区域跑起来

Terraform 骨架

# Define regions
variable "regions" {
  default = ["us-east-1", "eu-west-1", "ap-southeast-1"]
}

# Deploy worker fleet per region
module "captcha_workers" {
  for_each = toset(var.regions)
  source   = "./modules/captcha-worker"

  region              = each.key
  worker_count        = var.workers_per_region
  api_key_secret_arn  = aws_secretsmanager_secret.captchaai_key.arn
  task_queue_arn      = aws_sqs_queue.tasks[each.key].arn
  result_queue_arn    = aws_sqs_queue.results.arn
}

# SQS queue per region for task intake
resource "aws_sqs_queue" "tasks" {
  for_each = toset(var.regions)
  name     = "captcha-tasks-${each.key}"
}

# Central result queue
resource "aws_sqs_queue" "results" {
  name = "captcha-results-central"
}

Docker Compose（本地模拟多区域）

本地调试用不同 Redis 库号模拟隔离即可：

version: "3.8"
services:
  worker-us:
    build: ./worker
    environment:

      - CAPTCHAAI_API_KEY=${CAPTCHAAI_API_KEY}
      - WORKER_REGION=us-east-1
      - TASK_QUEUE=redis://redis:6379/0
    depends_on:

      - redis

  worker-eu:
    build: ./worker
    environment:

      - CAPTCHAAI_API_KEY=${CAPTCHAAI_API_KEY}
      - WORKER_REGION=eu-west-1
      - TASK_QUEUE=redis://redis:6379/1

  worker-ap:
    build: ./worker
    environment:

      - CAPTCHAAI_API_KEY=${CAPTCHAAI_API_KEY}
      - WORKER_REGION=ap-southeast-1
      - TASK_QUEUE=redis://redis:6379/2

  redis:
    image: redis:7-alpine

监控每个区域的健康状态

多区域最怕变慢了没人发现。定时探测端点，记录延迟和队列深度：

const axios = require("axios");

const REGIONS = ["us-east-1", "eu-west-1", "ap-southeast-1"];

async function checkRegionHealth() {
  const health = {};

  for (const region of REGIONS) {
    const endpoint = `https://${region}.workers.example.com/health`;
    try {
      const start = Date.now();
      const resp = await axios.get(endpoint, { timeout: 5000 });
      health[region] = {
        status: "healthy",
        latencyMs: Date.now() - start,
        activeWorkers: resp.data.activeWorkers,
        queueDepth: resp.data.queueDepth,
      };
    } catch (err) {
      health[region] = { status: "unhealthy", error: err.message };
    }
  }

  return health;
}

// Periodic health check
setInterval(async () => {
  const health = await checkRegionHealth();
  console.table(health);
}, 60000);

故障转移：自动摘除异常区域

某个区域挂了，任务要立刻改道，不能等人工介入：

def failover_check(region_health):
    """Redirect tasks from unhealthy regions."""
    healthy_regions = [
        r for r, h in region_health.items()
        if h["status"] == "healthy"
    ]

    if not healthy_regions:
        raise RuntimeError("All regions unhealthy")

    redirects = {}
    for region, health in region_health.items():
        if health["status"] == "unhealthy":
            # Pick the healthy region with lowest queue depth
            target = min(
                healthy_regions,
                key=lambda r: region_health[r].get("queue_depth", 0)
            )
            redirects[region] = target
            print(f"Failover: {region} → {target}")

    return redirects

故障转移的两条操作纪律

• 延迟、错误率或代理健康度超过阈值就停止往该区域路由新任务，别等它彻底不可用；正在进行的任务排空，而不是直接掐断。
• 只有该区域重新满足故障转移前用的同一套健康门槛，才恢复流量——门槛不能临时放宽。

常见问题

每个区域都要单独申请一个 CaptchaAI API Key 吗？

不需要。一个 API Key 全球通用，靠自己的监控指标按 region 字段拆分用量即可。

Worker 能不能部署在国内机房？

能运行，但目标站点在海外时不划算：既有到目标站点的网络延迟，部分验证码类型的外部脚本加载也不够稳定。更稳妥的做法是把 Worker 放到离目标站点更近的海外区域。

后续从三个区域扩展到更多区域，改动大吗？

不大。REGION_MAP、REGION_QUEUES 和 Terraform 的 regions 都是可扩展的映射/数组，加区域只是加一行配置。

多区域架构对 reCAPTCHA、Turnstile、GeeTest 等不同类型都要单独配置吗？

不用。method 参数指定具体类型，区域路由只管任务往哪送，同一套 Worker 代码处理所有类型。

成本：多花的是基础设施，不是 CaptchaAI

Worker 部署在哪个区域，CaptchaAI 收费都一样——多花的钱在云厂商账单上。

常见故障排查

相关文章

• CaptchaAI 负载均衡器架构
• CaptchaAI 验证码解决架构详解

下一步

获取你的 API Key，把 Worker 分散部署到多个区域。

相关指南：

• 每小时处理 10,000 个任务
• 按区域看识别性能
• Worker 自动扩缩容