单区域部署意味着单点故障，并且会增加远程目标的延迟。多区域架构将 CAPTCHA 工作人员部署在目标站点附近，减少 API 往返时间，并在区域中断时幸存下来。
架构概述
[Task Router]
(Route53 / Load Balancer)
↙ ↓ ↘
[US-East] [EU-West] [AP-Southeast]
Workers Workers Workers
↓ ↓ ↓
[CaptchaAI API] ← shared API key
↓ ↓ ↓
[Central DB / Queue]
(Results aggregation)
每个地区都有独立的工人。所有共享相同的 CaptchaAI API 密钥并将结果推送到中央存储。
当多区域很重要时
| 情况 | 单一区域 | 多区域 |
|---|---|---|
| 某一国家/地区的目标网站 | 充足的 | 矫枉过正 |
| 全球目标站点 | 增加了 100–300 毫秒的延迟 | 每个区域的本地延迟 |
| 99.9% 的正常运行时间要求 | 很难保证 | 自然冗余 |
| 监管数据驻留 | 无法遵守 | 本地处理 |
| < 1,000 个任务/hour | 美好的 | 不必要的复杂性 |
| > 10,000 个任务/hour | 缩放限制 | 分配负载 |
区域工人部署
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}
任务路由器
将任务路由到距离目标站点最近的区域：
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
基础设施设置
地形骨架
# 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（本地多区域模拟）
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
每个区域的健康监控
JavaScript
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
成本考虑
| 成分 | 成本因素 | 优化 |
|---|---|---|
| 工作实例 | 按区域计算 | 空闲时自动缩放至 0 |
| 跨区域数据传输 | 区域之间 $0.02/GB | 最小化结果有效负载大小 |
| SQS队列 | 按请求定价 | 尽可能批量发送消息 |
| CaptchaAI API | 不分地区，费用相同 | 无多地区保费 |
无论工作人员位于何处，CaptchaAI 均收取相同的费率 - 多区域成本仅是基础设施成本。
故障转移策略
- 当延迟、错误率或代理运行状况超过策略阈值时，停止将新任务路由到降级区域。
- 有意地排出正在进行的工作，而不是突然切换每个队列和会话。
- 仅在该区域满足故障转移前使用的相同运行状况门之后才恢复流量。
故障排除
| 问题 | 原因 | 处理方式 |
|---|---|---|
| 一个地区持续较慢 | 与 CaptchaAI 服务器的距离 | 比较基线延迟；可能是预期的 |
| 任务路由将所有内容发送到一个区域 | 基于域的路由太宽泛 | 添加更细粒度的路由规则 |
| 故障转移未触发 | 健康检查端点没有响应 | 确保运行状况端点位于与工作逻辑不同的路径上 |
| API 密钥余额消耗速度更快 | 所有区域共享一键 | 预期 - 监控总体使用情况 |
常问问题
每个区域是否需要单独的 CaptchaAI API 密钥？
不会。一个 API 密钥可在全球范围内使用。使用单个密钥并通过您自己的指标跟踪每个区域的使用情况。
实现高可用性的最少区域数是多少？
不同地理区域（例如美国 + 欧盟）的两个区域提供基本 HA。三个区域（美国+欧盟+亚洲）覆盖全球可用性。
我应该将工作人员部署在与 CaptchaAI 的服务器相同的区域吗？
CaptchaAI 的基础设施处理全球请求。将工作人员部署在您的“目标站点”附近以进行代理优化，而不是靠近 CaptchaAI。
相关文章
下一步
跨多个区域部署 CaptchaAI 工作人员 –”获取您的 API 密钥构建全球验证码解决基础设施。
相关指南：
