Lambda 验证码跟踪：DynamoDB 表设计实战

把验证码解决结果存到哪里，才不会被 Lambda 的连接池拖垮？答案是 DynamoDB——按需计费、没有连接数上限、内置 TTL 自动清理，天生适合无服务器架构。本文给出可直接复用的表结构设计、Python 与 JavaScript 完整实现，以及常用查询模式。

适用场景

典型场景是跨境电商爬虫团队：Lambda 每天并发处理数千次 reCAPTCHA v2 请求。

按 sitekey 回溯历史解决记录，定位异常站点
以 CaptchaAI 的 ADVANCE 套餐（$90/月，50 线程）支撑高并发，配合下面的单表设计落地

数据表设计

单表模式

一张 DynamoDB 表就能同时承担解决历史、进行中任务和每日统计三类数据：

分区键 (PK)	排序键 (SK)	用途
`SOLVE#{captcha_id}`	`META`	单次解决记录
`SITE#{sitekey}`	`SOLVE#{timestamp}`	按站点的解决历史
`STATS#{date}`	`TYPE#{captcha_type}`	按天聚合的统计数据
`ACTIVE#{captcha_id}`	`TASK`	进行中任务跟踪

建表定义

{
  "TableName": "CaptchaSolves",
  "KeySchema": [
    { "AttributeName": "PK", "KeyType": "HASH" },
    { "AttributeName": "SK", "KeyType": "RANGE" }
  ],
  "AttributeDefinitions": [
    { "AttributeName": "PK", "KeyType": "S" },
    { "AttributeName": "SK", "KeyType": "S" },
    { "AttributeName": "GSI1PK", "KeyType": "S" },
    { "AttributeName": "GSI1SK", "KeyType": "S" }
  ],
  "GlobalSecondaryIndexes": [
    {
      "IndexName": "GSI1",
      "KeySchema": [
        { "AttributeName": "GSI1PK", "KeyType": "HASH" },
        { "AttributeName": "GSI1SK", "KeyType": "RANGE" }
      ],
      "Projection": { "ProjectionType": "ALL" }
    }
  ],
  "BillingMode": "PAY_PER_REQUEST",
  "TimeToLiveSpecification": {
    "AttributeName": "ttl",
    "Enabled": true
  }
}

GSI1 可复用于跨类型状态查询，无需为不同前缀单独建表。

Python 实现方案

环境准备

import os
import time
from datetime import datetime, timezone
import boto3
import requests

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ.get("DYNAMODB_TABLE", "CaptchaSolves"))
API_KEY = os.environ["CAPTCHAAI_API_KEY"]

提交并跟踪解决状态

def solve_and_track(sitekey, pageurl, captcha_type="recaptcha_v2", project=None):
    now = datetime.now(timezone.utc)
    timestamp = now.isoformat()
    ttl_90_days = int(now.timestamp()) + (90 * 24 * 3600)

    # Submit to CaptchaAI
    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:
        # Store error record
        table.put_item(Item={
            "PK": f"SITE#{sitekey}",
            "SK": f"SOLVE#{timestamp}",
            "captcha_type": captcha_type,
            "pageurl": pageurl,
            "status": "error",
            "error": data.get("request"),
            "submitted_at": timestamp,
            "project": project or "default",
            "ttl": ttl_90_days,
            "GSI1PK": f"STATUS#error",
            "GSI1SK": timestamp
        })
        return {"error": data.get("request")}

    captcha_id = data["request"]

    # Track active task
    table.put_item(Item={
        "PK": f"ACTIVE#{captcha_id}",
        "SK": "TASK",
        "sitekey": sitekey,
        "pageurl": pageurl,
        "captcha_type": captcha_type,
        "submitted_at": timestamp,
        "ttl": int(now.timestamp()) + 600  # Auto-clean in 10 min
    })

    # Poll for result
    polls = 0
    for _ in range(60):
        time.sleep(5)
        polls += 1
        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:
            solved_at = datetime.now(timezone.utc).isoformat()
            elapsed_ms = int(
                (datetime.now(timezone.utc) - now).total_seconds() * 1000
            )

            # Store success record
            table.put_item(Item={
                "PK": f"SOLVE#{captcha_id}",
                "SK": "META",
                "captcha_type": captcha_type,
                "sitekey": sitekey,
                "pageurl": pageurl,
                "status": "solved",
                "submitted_at": timestamp,
                "solved_at": solved_at,
                "elapsed_ms": elapsed_ms,
                "polls": polls,
                "project": project or "default",
                "ttl": ttl_90_days,
                "GSI1PK": f"STATUS#solved",
                "GSI1SK": timestamp
            })

            # Also store in site history
            table.put_item(Item={
                "PK": f"SITE#{sitekey}",
                "SK": f"SOLVE#{timestamp}",
                "captcha_id": captcha_id,
                "status": "solved",
                "elapsed_ms": elapsed_ms,
                "ttl": ttl_90_days
            })

            # Remove active task
            table.delete_item(Key={
                "PK": f"ACTIVE#{captcha_id}", "SK": "TASK"
            })

            # Update daily stats
            update_daily_stats(captcha_type, True, elapsed_ms)

            return {"solution": result["request"]}

        if result.get("request") != "CAPCHA_NOT_READY":
            table.put_item(Item={
                "PK": f"SITE#{sitekey}",
                "SK": f"SOLVE#{timestamp}",
                "captcha_id": captcha_id,
                "status": "error",
                "error": result.get("request"),
                "ttl": ttl_90_days
            })
            table.delete_item(Key={
                "PK": f"ACTIVE#{captcha_id}", "SK": "TASK"
            })
            update_daily_stats(captcha_type, False, 0)
            return {"error": result.get("request")}

    table.delete_item(Key={"PK": f"ACTIVE#{captcha_id}", "SK": "TASK"})
    update_daily_stats(captcha_type, False, 0)
    return {"error": "TIMEOUT"}


def update_daily_stats(captcha_type, success, elapsed_ms):
    date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    update_expr = "SET total_solves = if_not_exists(total_solves, :zero) + :one"
    expr_values = {":zero": 0, ":one": 1}

    if success:
        update_expr += ", successful = if_not_exists(successful, :zero) + :one"
        update_expr += ", total_elapsed = if_not_exists(total_elapsed, :zero) + :elapsed"
        expr_values[":elapsed"] = elapsed_ms
    else:
        update_expr += ", failed = if_not_exists(failed, :zero) + :one"

    table.update_item(
        Key={"PK": f"STATS#{date_str}", "SK": f"TYPE#{captcha_type}"},
        UpdateExpression=update_expr,
        ExpressionAttributeValues=expr_values
    )

常用查询模式

def get_site_history(sitekey, limit=50):
    """Get recent solves for a specific site key."""
    response = table.query(
        KeyConditionExpression="PK = :pk",
        ExpressionAttributeValues={":pk": f"SITE#{sitekey}"},
        ScanIndexForward=False,
        Limit=limit
    )
    return response["Items"]


def get_daily_stats(date_str=None):
    """Get stats for a specific date (default: today)."""
    if not date_str:
        date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")

    response = table.query(
        KeyConditionExpression="PK = :pk",
        ExpressionAttributeValues={":pk": f"STATS#{date_str}"}
    )
    return response["Items"]


def get_active_tasks():
    """List all currently active CAPTCHA tasks."""
    response = table.query(
        IndexName="GSI1",
        KeyConditionExpression="GSI1PK = :pk",
        ExpressionAttributeValues={":pk": "STATUS#polling"}
    )
    return response["Items"]

JavaScript 实现方案

const { DynamoDBClient } = require("@aws-sdk/client-dynamodb");
const { DynamoDBDocumentClient, PutCommand, QueryCommand, UpdateCommand } = require("@aws-sdk/lib-dynamodb");
const axios = require("axios");

const client = DynamoDBDocumentClient.from(new DynamoDBClient({}));
const TABLE = process.env.DYNAMODB_TABLE || "CaptchaSolves";
const API_KEY = process.env.CAPTCHAAI_API_KEY;

async function solveAndTrack(sitekey, pageurl, type = "recaptcha_v2") {
  const now = new Date();
  const timestamp = now.toISOString();
  const ttl = Math.floor(now.getTime() / 1000) + 90 * 24 * 3600;

  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) {
    await client.send(new PutCommand({
      TableName: TABLE,
      Item: { PK: `SITE#${sitekey}`, SK: `SOLVE#${timestamp}`, status: "error", error: submit.data.request, ttl },
    }));
    return { error: submit.data.request };
  }

  const captchaId = submit.data.request;
  let polls = 0;

  for (let i = 0; i < 60; i++) {
    await new Promise((r) => setTimeout(r, 5000));
    polls++;
    const poll = await axios.get("https://ocr.captchaai.com/res.php", {
      params: { key: API_KEY, action: "get", id: captchaId, json: 1 },
    });

    if (poll.data.status === 1) {
      const elapsed = Date.now() - now.getTime();
      await client.send(new PutCommand({
        TableName: TABLE,
        Item: {
          PK: `SOLVE#${captchaId}`, SK: "META", captcha_type: type,
          sitekey, pageurl, status: "solved", submitted_at: timestamp,
          solved_at: new Date().toISOString(), elapsed_ms: elapsed, polls, ttl,
        },
      }));
      return { solution: poll.data.request };
    }

    if (poll.data.request !== "CAPCHA_NOT_READY") {
      return { error: poll.data.request };
    }
  }
  return { error: "TIMEOUT" };
}

async function getSiteHistory(sitekey, limit = 50) {
  const result = await client.send(new QueryCommand({
    TableName: TABLE,
    KeyConditionExpression: "PK = :pk",
    ExpressionAttributeValues: { ":pk": `SITE#${sitekey}` },
    ScanIndexForward: false,
    Limit: limit,
  }));
  return result.Items;
}

成本优化

可变负载采用按需计费 - 避免过度预置容量
开启 TTL 自动清理过期记录 - 降低存储成本
查询时只投影需要的属性 - 减少读取单元消耗
用 BatchWriteItem 批量写入 - 减少 API 调用次数
用 DynamoDB Streams 做分析 - 把聚合计算卸载给 Lambda

故障排除

ProvisionedThroughputExceededException - 每秒写入次数超限，切换按需计费或提高预置 WCU
TTL 项目没有立即删除 - TTL 删除是最终一致的（约 48 小时内），查询时自行过滤过期项目
STATS#{date} 出现热分区 - 所有 worker 写同一个分区，加随机后缀分片 STATS#{date}#shard{0-9}
查询返回的项目太多 - 分区键设计太宽，补充 SK 条件缩小结果范围

常见问题

无服务器场景下，为什么选 DynamoDB 而不是 RDS？

Lambda 每次调用都可能新开连接，RDS 容易被打满，得接 RDS Proxy。DynamoDB 没有连接数上限，适配这种短生命周期调用。

多个项目共用一张表，记录会互相覆盖吗？

不会。project 字段区分统计维度，配合 SITE#{sitekey} 分区键，不同项目的解决记录天然隔离，不需要单独建表。

每天 10,000 次解决量，DynamoDB 费用大概多少？

每百万次写入约 $1.25，读取约 $0.25。每天 10,000 次解决，月成本通常不到 $1。

TTL 到期的记录会立刻消失吗？

不会。TTL 清理最终一致，约 48 小时内完成，查询时要自行过滤已过期项目。

下一步

获取你的 CaptchaAI API 密钥，把上面的表结构接入现有的 Lambda 函数，搭建能随流量自动伸缩的验证码跟踪方案。

相关指南：

用于无服务器验证码解决跟踪的 DynamoDB

适用场景

数据表设计

单表模式

建表定义

Python 实现方案

环境准备

提交并跟踪解决状态

常用查询模式

JavaScript 实现方案

成本优化

故障排除

常见问题

无服务器场景下，为什么选 DynamoDB 而不是 RDS？

多个项目共用一张表，记录会互相覆盖吗？

每天 10,000 次解决量，DynamoDB 费用大概多少？

TTL 到期的记录会立刻消失吗？

下一步

AWS Lambda + CaptchaAI：无服务器验证码解决

用于 CaptchaAI Worker 部署的 Ansible Playbook

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

验证码解决基础设施的蓝绿部署

Azure Functions + CaptchaAI：云集成

构建 CaptchaAI 使用仪表板和监控

适用场景

数据表设计

单表模式

建表定义

Python 实现方案

环境准备

提交并跟踪解决状态

常用查询模式

JavaScript 实现方案

成本优化

故障排除

常见问题

无服务器场景下，为什么选 DynamoDB 而不是 RDS？

多个项目共用一张表，记录会互相覆盖吗？

每天 10,000 次解决量，DynamoDB 费用大概多少？

TTL 到期的记录会立刻消失吗？

下一步

相关文章

AWS Lambda + CaptchaAI：无服务器验证码解决

用于 CaptchaAI Worker 部署的 Ansible Playbook

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

验证码解决基础设施的蓝绿部署

Azure Functions + CaptchaAI：云集成

构建 CaptchaAI 使用仪表板和监控