验证码去重方案：Redis 锁与 PostgreSQL 咨询锁

同一个验证码被多个 worker 或重试逻辑重复提交时，你在为同一次识别反复付费。4 台服务器同时对接同一个 reCAPTCHA v3 sitekey，网络抖动引发的重试就能让请求量翻两三倍——一层几十行代码的去重逻辑就能省下这笔钱：拦下重复请求，直接返回已有结果。

重复请求都是怎么来的

常见原因有以下几种：

结果还没返回就重试——重试逻辑过于激进，成本可能翻 2-5 倍。
多个 worker 打同一目标——worker 之间没有协调，导致并行重复求解。
页面刷新重新触发——前端超时后自动重试，每次刷新都多算一次。
队列消息重复投递——at-least-once 语义下，每次重放都要重新求解。

去重键该怎么设计

去重的核心是从请求参数生成一个唯一键：

import hashlib


def dedup_key(method, sitekey, pageurl):
    """Generate a deduplication key for a CAPTCHA solve request."""
    raw = f"{method}:{sitekey}:{pageurl}"
    return f"captcha:dedup:{hashlib.sha256(raw.encode()).hexdigest()[:16]}"

提示： 键里只放能决定请求是否相同的字段，代理、User-Agent 等无关字段不要放入。下表是各类型的键组成：

验证码类型	关键组成
reCAPTCHA v2	`method` + `sitekey` + `pageurl`
reCAPTCHA v3	`method` + `sitekey` + `pageurl` + `action`
hCaptcha	`method` + `sitekey` + `pageurl`
Turnstile	`method` + `sitekey` + `pageurl`
图片验证码	`method` + `body` 内容哈希

用 Redis 实现去重

Redis 的 TTL 和原子写操作天然适合做去重锁，下面是可以直接接入生产环境的完整实现。

Python 实现

逻辑很直接：键存在就等待或读缓存，不存在就标记为 solving 并调用 CaptchaAI。

import os
import time
import json
import hashlib
import redis
import requests

r = redis.Redis(
    host=os.environ.get("REDIS_HOST", "localhost"),
    port=int(os.environ.get("REDIS_PORT", 6379)),
    decode_responses=True
)

API_KEY = os.environ["CAPTCHAAI_API_KEY"]

# Dedup window: how long to consider a request "in progress"
DEDUP_TTL = 180  # seconds


def dedup_key(method, sitekey, pageurl, extra=""):
    raw = f"{method}:{sitekey}:{pageurl}:{extra}"
    return f"captcha:dedup:{hashlib.sha256(raw.encode()).hexdigest()[:16]}"


def solve_with_dedup(sitekey, pageurl, method="userrecaptcha"):
    key = dedup_key(method, sitekey, pageurl)

    # Check if this request is already being solved
    existing = r.get(key)
    if existing:
        state = json.loads(existing)
        if state["status"] == "solving":
            # Wait for the result
            return wait_for_result(key)
        elif state["status"] == "solved":
            return {"solution": state["solution"], "source": "dedup_cache"}
        elif state["status"] == "error":
            pass  # Allow retry on error

    # Mark as solving
    r.set(key, json.dumps({"status": "solving", "started": time.time()}), ex=DEDUP_TTL)

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

    if data.get("status") != 1:
        r.set(key, json.dumps({"status": "error", "error": data.get("request")}), ex=30)
        return {"error": data.get("request")}

    captcha_id = data["request"]

    # Poll for result
    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:
            solution = result["request"]
            # Cache the result for other workers (short TTL since tokens expire)
            r.set(key, json.dumps({
                "status": "solved",
                "solution": solution,
                "solved_at": time.time()
            }), ex=60)  # Cache result for 60 seconds
            return {"solution": solution, "source": "api"}

        if result.get("request") != "CAPCHA_NOT_READY":
            r.set(key, json.dumps({
                "status": "error", "error": result.get("request")
            }), ex=30)
            return {"error": result.get("request")}

    r.set(key, json.dumps({"status": "error", "error": "TIMEOUT"}), ex=30)
    return {"error": "TIMEOUT"}


def wait_for_result(key, timeout=120):
    """Wait for another worker to finish solving."""
    start = time.time()
    while time.time() - start < timeout:
        data = r.get(key)
        if data:
            state = json.loads(data)
            if state["status"] == "solved":
                return {"solution": state["solution"], "source": "dedup_wait"}
            if state["status"] == "error":
                return {"error": state.get("error", "UNKNOWN")}
        time.sleep(2)
    return {"error": "DEDUP_WAIT_TIMEOUT"}

JavaScript 实现

Node.js 版本的逻辑完全一致，用 ioredis 和 axios 实现：

const Redis = require("ioredis");
const axios = require("axios");
const crypto = require("crypto");

const redis = new Redis(process.env.REDIS_URL || "redis://localhost:6379");
const API_KEY = process.env.CAPTCHAAI_API_KEY;
const DEDUP_TTL = 180;

function dedupKey(method, sitekey, pageurl) {
  const raw = `${method}:${sitekey}:${pageurl}`;
  const hash = crypto.createHash("sha256").update(raw).digest("hex").slice(0, 16);
  return `captcha:dedup:${hash}`;
}

async function solveWithDedup(sitekey, pageurl, method = "userrecaptcha") {
  const key = dedupKey(method, sitekey, pageurl);

  // Check existing
  const existing = await redis.get(key);
  if (existing) {
    const state = JSON.parse(existing);
    if (state.status === "solving") return await waitForResult(key);
    if (state.status === "solved") return { solution: state.solution, source: "dedup_cache" };
  }

  // Mark as solving
  await redis.set(key, JSON.stringify({ status: "solving", started: Date.now() }), "EX", DEDUP_TTL);

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

  if (submit.data.status !== 1) {
    await redis.set(key, JSON.stringify({ status: "error", error: submit.data.request }), "EX", 30);
    return { error: submit.data.request };
  }

  const captchaId = submit.data.request;

  for (let i = 0; i < 60; 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: captchaId, json: 1 },
    });

    if (poll.data.status === 1) {
      await redis.set(key, JSON.stringify({ status: "solved", solution: poll.data.request }), "EX", 60);
      return { solution: poll.data.request, source: "api" };
    }
    if (poll.data.request !== "CAPCHA_NOT_READY") {
      await redis.set(key, JSON.stringify({ status: "error", error: poll.data.request }), "EX", 30);
      return { error: poll.data.request };
    }
  }

  await redis.set(key, JSON.stringify({ status: "error", error: "TIMEOUT" }), "EX", 30);
  return { error: "TIMEOUT" };
}

async function waitForResult(key, timeout = 120000) {
  const start = Date.now();
  while (Date.now() - start < timeout) {
    const data = await redis.get(key);
    if (data) {
      const state = JSON.parse(data);
      if (state.status === "solved") return { solution: state.solution, source: "dedup_wait" };
      if (state.status === "error") return { error: state.error };
    }
    await new Promise((r) => setTimeout(r, 2000));
  }
  return { error: "DEDUP_WAIT_TIMEOUT" };
}

没有 Redis？用 PostgreSQL 数据库锁

不想为去重单独引入 Redis 依赖？PostgreSQL 的咨询锁（advisory lock）能做同样的事：抢不到锁的 worker 直接阻塞等待，锁一释放就去缓存表里读结果。

import psycopg2


def solve_with_pg_dedup(conn, sitekey, pageurl):
    """Use PostgreSQL advisory locks for deduplication."""
    # Generate a numeric lock key from the dedup key
    lock_id = hash(f"{sitekey}:{pageurl}") & 0x7FFFFFFF

    cursor = conn.cursor()

    # Try to acquire advisory lock (non-blocking)
    cursor.execute("SELECT pg_try_advisory_lock(%s)", (lock_id,))
    acquired = cursor.fetchone()[0]

    if not acquired:
        # Another worker is solving — wait for result
        cursor.execute("SELECT pg_advisory_lock(%s)", (lock_id,))
        # Lock acquired means other worker finished — check cache
        cursor.execute(
            "SELECT solution FROM captcha_cache "
            "WHERE sitekey = %s AND pageurl = %s "
            "AND created_at > NOW() - INTERVAL '60 seconds'",
            (sitekey, pageurl)
        )
        row = cursor.fetchone()
        cursor.execute("SELECT pg_advisory_unlock(%s)", (lock_id,))
        if row:
            return {"solution": row[0], "source": "pg_cache"}
        return {"error": "NO_CACHED_RESULT"}

    try:
        # Solve the CAPTCHA
        solution = solve_via_api(sitekey, pageurl)
        if solution:
            cursor.execute(
                "INSERT INTO captcha_cache (sitekey, pageurl, solution) "
                "VALUES (%s, %s, %s)",
                (sitekey, pageurl, solution)
            )
            conn.commit()
        return {"solution": solution} if solution else {"error": "SOLVE_FAILED"}
    finally:
        cursor.execute("SELECT pg_advisory_unlock(%s)", (lock_id,))

怎么衡量去重效果

上线之后要能拿出数字证明去重真的省钱了：

记录哪些计数

def track_dedup_stats(source):
    """Increment counters for dedup tracking."""
    today = time.strftime("%Y-%m-%d")
    r.hincrby(f"dedup:stats:{today}", source, 1)
    r.expire(f"dedup:stats:{today}", 7 * 86400)


def get_dedup_report():
    today = time.strftime("%Y-%m-%d")
    stats = r.hgetall(f"dedup:stats:{today}")
    total = sum(int(v) for v in stats.values())
    saved = int(stats.get("dedup_cache", 0)) + int(stats.get("dedup_wait", 0))
    return {
        "total_requests": total,
        "deduplicated": saved,
        "savings_pct": f"{saved / total * 100:.1f}%" if total else "0%",
        "breakdown": stats
    }

常见故障排查

最常踩的坑：

去重键冲突（哈希太短或漏参数）——纳入全部相关参数，加长哈希。
等待方超时（求解 worker 崩溃）——solving TTL 自动过期（180 秒），无需人工干预。
缓存结果过期（结果 TTL 比 token 有效期长）——结果 TTL 要短于 token 生命周期（约 60 秒）。
两个 worker 抢锁（检查和加锁非原子）——用 SET NX 保证原子性。
换代理后去重失效（代理被纳入了键）——去重键不含代理，token 与代理无关。

常见问题

去重和限流是一回事吗？

不是。去重解决“重复请求”，限流解决“速率超限”，两者各管一段，通常一起用。

什么情况下值得引入去重层？

多个 worker 面对同一个 sitekey/pageurl 组合时就值得。去重率哪怕只有 10%，量大时也能省下不少 API 额度。

去重键的 TTL 应该设多长？

“正在求解”状态建议 60-180 秒；“已求解”结果的缓存要更短，须短于 token 有效期（reCAPTCHA 约 60 秒）。

图片验证码需要去重吗？

需要，但去重键要用图片内容哈希而非 URL——相同图片答案相同，只用 URL 会让换个链接就重新求解一次。

下一步

从今天开始不用再为重复请求付费——获取 CaptchaAI API 密钥，接入上面这套去重层。

相关阅读：

通过数据库锁定消除重复验证码解决请求

重复请求都是怎么来的

去重键该怎么设计

用 Redis 实现去重

Python 实现

JavaScript 实现

没有 Redis？用 PostgreSQL 数据库锁

怎么衡量去重效果

记录哪些计数

常见故障排查

常见问题

去重和限流是一回事吗？

什么情况下值得引入去重层？

去重键的 TTL 应该设多长？

图片验证码需要去重吗？

下一步

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

用于 CaptchaAI Worker 部署的 Ansible Playbook

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

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

Azure Functions + CaptchaAI：云集成

构建 CaptchaAI 使用仪表板和监控

重复请求都是怎么来的

去重键该怎么设计

用 Redis 实现去重

Python 实现

JavaScript 实现

没有 Redis？用 PostgreSQL 数据库锁

怎么衡量去重效果

记录哪些计数

常见故障排查

常见问题

去重和限流是一回事吗？

什么情况下值得引入去重层？

去重键的 TTL 应该设多长？

图片验证码需要去重吗？

下一步

相关文章

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

用于 CaptchaAI Worker 部署的 Ansible Playbook

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

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

Azure Functions + CaptchaAI：云集成

构建 CaptchaAI 使用仪表板和监控