使用 AWS SNS 和 CaptchaAI 构建事件驱动的验证码解决方案

爬虫是不是也在用 while 循环，每隔几秒问一次 res.php“结果出来了吗”？轮询占线程，还焊死了抓取器和识别管道。更省心的做法是反过来：CaptchaAI 识别完成后主动回调，发布到 AWS SNS，下游各自独立处理。

本文会带你搭出这条链路：

• Lambda 接回调
• SNS 扇出
• SQS 消费 + 排错

架构概览：从轮询变成事件驱动

整体链路长这样：

[Scraper] → Submit CAPTCHA → [CaptchaAI API]
                                    ↓
                            Solve completes
                                    ↓
                            Callback → [API Gateway + Lambda]
                                    ↓
                            Publish → [SNS Topic]
                                    ↓
                    ┌───────────────┼───────────────┐
                    ↓               ↓               ↓
            [SQS Queue]      [Lambda Logger]   [Email Alert]
            (result store)   (audit trail)     (on failure)

关键在扇出（fan-out）：

1. 一次结果同时触发多个消费者。
2. 加消费者只需新增订阅，不改回调代码。
3. 单个下游故障不拖垮整条链路。

第一步：创建 SNS 主题

AWS CLI 或 boto3 建主题都行。

AWS CLI

aws sns create-topic --name captcha-results --output text
# Returns: arn:aws:sns:us-east-1:123456789:captcha-results

Python（boto3）

import boto3

sns = boto3.client("sns", region_name="us-east-1")

response = sns.create_topic(Name="captcha-results")
topic_arn = response["TopicArn"]
print(f"Topic ARN: {topic_arn}")

国内装 boto3 可加清华镜像加速。

第二步：搭建回调接收器

接住回调，解析 task_id 和结果，发布到 SNS。

Python（Lambda 处理程序）

import json
import os
import boto3

sns = boto3.client("sns")
TOPIC_ARN = os.environ["SNS_TOPIC_ARN"]


def lambda_handler(event, context):
    """Receive CaptchaAI callback and publish to SNS."""
    # Parse query parameters from API Gateway
    params = event.get("queryStringParameters", {}) or {}
    task_id = params.get("id", "")
    solution = params.get("code", "")

    if not task_id or not solution:
        return {"statusCode": 400, "body": "Missing id or code"}

    # Publish to SNS
    message = {
        "task_id": task_id,
        "solution": solution,
        "status": "solved"
    }

    sns.publish(
        TopicArn=TOPIC_ARN,
        Message=json.dumps(message),
        Subject="captcha-solved",
        MessageAttributes={
            "task_id": {
                "DataType": "String",
                "StringValue": task_id
            }
        }
    )

    return {"statusCode": 200, "body": "OK"}

JavaScript（Lambda 处理程序）

const { SNSClient, PublishCommand } = require("@aws-sdk/client-sns");

const sns = new SNSClient({ region: "us-east-1" });
const TOPIC_ARN = process.env.SNS_TOPIC_ARN;

exports.handler = async (event) => {
  const params = event.queryStringParameters || {};
  const taskId = params.id;
  const solution = params.code;

  if (!taskId || !solution) {
    return { statusCode: 400, body: "Missing id or code" };
  }

  const message = {
    task_id: taskId,
    solution: solution,
    status: "solved",
  };

  await sns.send(
    new PublishCommand({
      TopicArn: TOPIC_ARN,
      Message: JSON.stringify(message),
      Subject: "captcha-solved",
      MessageAttributes: {
        task_id: { DataType: "String", StringValue: taskId },
      },
    })
  );

  return { statusCode: 200, body: "OK" };
};

提示：MessageAttributes 带上 task_id，方便下游过滤或幂等校验。

第三步：提交验证码时带上回调地址

两个改动：

• pingback 指向刚才的 API Gateway 端点
• 提交后不再轮询，等回调自动触发

Python

import os
import requests

API_KEY = os.environ["CAPTCHAAI_API_KEY"]
CALLBACK_URL = os.environ["CALLBACK_GATEWAY_URL"]  # API Gateway URL


def submit_captcha(sitekey, pageurl):
    """Submit CAPTCHA with SNS-backed callback."""
    resp = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "pingback": CALLBACK_URL,
        "json": 1
    })
    data = resp.json()

    if data.get("status") == 1:
        return data["request"]  # task_id
    raise RuntimeError(f"Submit failed: {data.get('request')}")

第四步：订阅下游消费者

一个 SNS 主题可挂多种协议的订阅者，常见三种：

1. SQS 队列——落地存储结果。
2. Lambda——审计日志。
3. 邮件——失败告警。

SQS 队列（落地存储结果）

# Subscribe an SQS queue to receive all results
sqs_arn = "arn:aws:sqs:us-east-1:123456789:captcha-results-queue"

sns.subscribe(
    TopicArn=topic_arn,
    Protocol="sqs",
    Endpoint=sqs_arn
)

Lambda（审计日志）

# Subscribe a Lambda for audit logging
lambda_arn = "arn:aws:lambda:us-east-1:123456789:function:captcha-audit-logger"

sns.subscribe(
    TopicArn=topic_arn,
    Protocol="lambda",
    Endpoint=lambda_arn
)

邮件（失败告警）

# Subscribe email for error notifications with filter
sns.subscribe(
    TopicArn=topic_arn,
    Protocol="email",
    Endpoint="[email protected]"
)

第五步：从 SQS 取识别结果

爬虫改从 SQS 长轮询拿结果，不再轮询 CaptchaAI。

Python

import json
import boto3

sqs = boto3.client("sqs", region_name="us-east-1")
QUEUE_URL = os.environ["SQS_QUEUE_URL"]


def get_solved_captcha(timeout=30):
    """Wait for a CAPTCHA solution from the SQS queue."""
    response = sqs.receive_message(
        QueueUrl=QUEUE_URL,
        MaxNumberOfMessages=1,
        WaitTimeSeconds=min(timeout, 20)  # Long polling (max 20s)
    )

    messages = response.get("Messages", [])
    if not messages:
        return None

    msg = messages[0]
    # SNS wraps the message — unwrap it
    sns_envelope = json.loads(msg["Body"])
    result = json.loads(sns_envelope["Message"])

    # Delete message after processing
    sqs.delete_message(
        QueueUrl=QUEUE_URL,
        ReceiptHandle=msg["ReceiptHandle"]
    )

    return result

JavaScript

const {
  SQSClient,
  ReceiveMessageCommand,
  DeleteMessageCommand,
} = require("@aws-sdk/client-sqs");

const sqs = new SQSClient({ region: "us-east-1" });
const QUEUE_URL = process.env.SQS_QUEUE_URL;

async function getSolvedCaptcha(timeout = 30) {
  const response = await sqs.send(
    new ReceiveMessageCommand({
      QueueUrl: QUEUE_URL,
      MaxNumberOfMessages: 1,
      WaitTimeSeconds: Math.min(timeout, 20),
    })
  );

  const messages = response.Messages || [];
  if (messages.length === 0) return null;

  const msg = messages[0];
  const snsEnvelope = JSON.parse(msg.Body);
  const result = JSON.parse(snsEnvelope.Message);

  await sqs.send(
    new DeleteMessageCommand({
      QueueUrl: QUEUE_URL,
      ReceiptHandle: msg.ReceiptHandle,
    })
  );

  return result;
}

常见故障排查

按这个顺序定位：

1. API Gateway 日志——回调到没到。
2. SNS 投递指标——发布成没成功。
3. 订阅者自身日志——消费端逻辑有没有问题。

进阶：按状态过滤 SNS 消息

失败告警不想混进正常队列？加条 FilterPolicy 就够。

# Only send failures to the ops queue
sns.subscribe(
    TopicArn=topic_arn,
    Protocol="sqs",
    Endpoint=failure_queue_arn,
    Attributes={
        "FilterPolicy": json.dumps({
            "status": ["failed", "error"]
        })
    }
)

常见问题

国内团队用 AWS SNS 稳定吗？该选哪个区域？

主题和 Lambda 选近数据源的区域（香港 ap-east-1、新加坡 ap-southeast-1），比 us-east-1 抖动小。

为什么不直接在回调里处理结果？

不接 SNS，每加需求都要改代码；接了就不用碰回调 Lambda。

为什么同一个识别结果会被处理两次？

SNS/SQS 都是至少一次投递——消费端做幂等，先查 task_id。

结果顺序乱了怎么办？可以用 SNS FIFO 吗？

可以，换 SNS FIFO 主题 + SQS FIFO 队列，MessageGroupId 设成 task_id。

验证码量不大，还有必要上 SNS 吗？

量小、单消费者轮询更简单；多路分发再切事件驱动。

相关阅读

• 构建客户端验证码处理管道
• 搭建合规的自动化流程
• 搭建 CaptchaAI 用量监控看板

开始动手

注册 CaptchaAI 拿 API Key，接入 AWS 管道。

延伸阅读：

• AWS Lambda + CaptchaAI 无服务器集成
• 回调地址与 Webhook 配置指南
• Webhook 安全：如何校验回调请求