CaptchaAI 回调失败处理：重试机制与死信队列实战

CaptchaAI 用回调（pingback）把识别结果推送到你的服务器，省去轮询开销。但只要服务器宕机、返回 5xx，或者网络超时，这条推送就可能悄悄丢失——业务逻辑对此毫不知情。本文给出三种可叠加使用的处理模式：兜底轮询、死信队列、幂等处理，代码可直接拿来改。

回调为什么会悄悄丢结果

服务器宕机——请求被直接拒绝，结果送不到你手上。
服务器返回 5xx——是否重试要看对方实现。
网络超时——连接长时间挂起，结果可能永久丢失。
处理程序崩溃——请求已接受但未落库，结果被静默丢弃。

永远不要只依赖回调本身——生产环境必须有兜底手段。

模式一：回调 + 兜底轮询双保险

最稳妥的做法是双保险：回调到达就直接用；没能在超时时间内收到回调的任务，交给后台轮询兜底捞回来。

Python

import os
import time
import threading
import requests
from flask import Flask, request

app = Flask(__name__)
API_KEY = os.environ["CAPTCHAAI_API_KEY"]

# Track task state
pending_tasks = {}  # task_id -> {"submitted_at": timestamp, "status": "pending"}
results = {}
lock = threading.Lock()


def submit_captcha(sitekey, pageurl, callback_url):
    """Submit with callback, but track for fallback polling."""
    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:
        task_id = data["request"]
        with lock:
            pending_tasks[task_id] = {
                "submitted_at": time.time(),
                "status": "pending"
            }
        return task_id
    return None


@app.route("/callback")
def captcha_callback():
    """Primary result delivery — CaptchaAI sends results here."""
    task_id = request.args.get("id")
    solution = request.args.get("code")

    with lock:
        results[task_id] = solution
        pending_tasks.pop(task_id, None)

    return "OK", 200


def fallback_poller():
    """Poll for any tasks that missed their callback."""
    while True:
        time.sleep(30)  # Check every 30 seconds

        with lock:
            stale_tasks = [
                tid for tid, info in pending_tasks.items()
                if time.time() - info["submitted_at"] > 120  # 2 min callback timeout
                and info["status"] == "pending"
            ]

        for task_id in stale_tasks:
            resp = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": API_KEY,
                "action": "get",
                "id": task_id,
                "json": 1
            })
            data = resp.json()

            if data.get("status") == 1:
                with lock:
                    results[task_id] = data["request"]
                    pending_tasks.pop(task_id, None)
                print(f"Fallback poll recovered: {task_id}")
            elif data.get("request") != "CAPCHA_NOT_READY":
                # Permanent error — remove from pending
                with lock:
                    pending_tasks.pop(task_id, None)
                print(f"Task failed: {task_id} — {data.get('request')}")


# Start fallback poller in background
poller_thread = threading.Thread(target=fallback_poller, daemon=True)
poller_thread.start()

JavaScript

const express = require("express");
const axios = require("axios");

const app = express();
const API_KEY = process.env.CAPTCHAAI_API_KEY;

const pendingTasks = new Map(); // taskId -> { submittedAt, status }
const results = new Map();

async function submitCaptcha(sitekey, pageurl, callbackUrl) {
  const resp = await axios.post("https://ocr.captchaai.com/in.php", null, {
    params: {
      key: API_KEY,
      method: "userrecaptcha",
      googlekey: sitekey,
      pageurl: pageurl,
      pingback: callbackUrl,
      json: 1,
    },
  });

  if (resp.data.status === 1) {
    const taskId = resp.data.request;
    pendingTasks.set(taskId, {
      submittedAt: Date.now(),
      status: "pending",
    });
    return taskId;
  }
  return null;
}

// Primary callback endpoint
app.get("/callback", (req, res) => {
  const taskId = req.query.id;
  const solution = req.query.code;

  results.set(taskId, solution);
  pendingTasks.delete(taskId);

  res.sendStatus(200);
});

// Fallback poller
setInterval(async () => {
  const now = Date.now();
  const staleTasks = [];

  for (const [taskId, info] of pendingTasks) {
    if (now - info.submittedAt > 120000 && info.status === "pending") {
      staleTasks.push(taskId);
    }
  }

  for (const taskId of staleTasks) {
    try {
      const resp = await axios.get("https://ocr.captchaai.com/res.php", {
        params: { key: API_KEY, action: "get", id: taskId, json: 1 },
      });

      if (resp.data.status === 1) {
        results.set(taskId, resp.data.request);
        pendingTasks.delete(taskId);
        console.log(`Fallback recovered: ${taskId}`);
      } else if (resp.data.request !== "CAPCHA_NOT_READY") {
        pendingTasks.delete(taskId);
        console.log(`Task failed: ${taskId} — ${resp.data.request}`);
      }
    } catch (err) {
      console.error(`Poll error for ${taskId}: ${err.message}`);
    }
  }
}, 30000);

app.listen(3000);

部署提示：回调接收端在国内时跨境延迟通常更高，把 120 秒的兜底阈值调宽到 180–240 秒可减少误判；轮询不额外计费，CaptchaAI 按线程数计费。

模式二：死信队列——失败的结果不能就地消失

回调处理程序处理结果时如果自己出错（数据库连不上、字段校验失败），不能让结果直接没了——先落到死信队列，等问题修复后补处理。618、双 11 大促期间数据库压力最大，正是死信队列最常派上用场的时候。

Python

import json
import os
import time
from pathlib import Path

DEAD_LETTER_DIR = Path("dead_letter")
DEAD_LETTER_DIR.mkdir(exist_ok=True)


@app.route("/callback")
def captcha_callback_with_dlq():
    task_id = request.args.get("id")
    solution = request.args.get("code")

    try:
        # Attempt normal processing
        store_result(task_id, solution)
        return "OK", 200
    except Exception as e:
        # Processing failed — save to dead-letter queue
        dead_letter = {
            "task_id": task_id,
            "solution": solution,
            "error": str(e),
            "received_at": time.time()
        }
        dlq_path = DEAD_LETTER_DIR / f"{task_id}.json"
        dlq_path.write_text(json.dumps(dead_letter))

        print(f"DLQ: {task_id} — {e}")
        return "OK", 200  # Still return 200 to CaptchaAI


def reprocess_dead_letters():
    """Retry processing dead-letter items."""
    for dlq_file in DEAD_LETTER_DIR.glob("*.json"):
        item = json.loads(dlq_file.read_text())

        try:
            store_result(item["task_id"], item["solution"])
            dlq_file.unlink()  # Remove after successful processing
            print(f"DLQ reprocessed: {item['task_id']}")
        except Exception:
            pass  # Leave in DLQ for next retry

JavaScript

const fs = require("fs");
const path = require("path");

const DLQ_DIR = path.join(__dirname, "dead_letter");
if (!fs.existsSync(DLQ_DIR)) fs.mkdirSync(DLQ_DIR);

app.get("/callback-dlq", (req, res) => {
  const taskId = req.query.id;
  const solution = req.query.code;

  try {
    storeResult(taskId, solution);
    res.sendStatus(200);
  } catch (err) {
    // Save to dead-letter queue
    const deadLetter = {
      task_id: taskId,
      solution: solution,
      error: err.message,
      received_at: Date.now(),
    };

    fs.writeFileSync(
      path.join(DLQ_DIR, `${taskId}.json`),
      JSON.stringify(deadLetter)
    );

    console.log(`DLQ: ${taskId} — ${err.message}`);
    res.sendStatus(200); // Still acknowledge to CaptchaAI
  }
});

function reprocessDeadLetters() {
  const files = fs.readdirSync(DLQ_DIR).filter((f) => f.endsWith(".json"));

  for (const file of files) {
    const filePath = path.join(DLQ_DIR, file);
    const item = JSON.parse(fs.readFileSync(filePath, "utf8"));

    try {
      storeResult(item.task_id, item.solution);
      fs.unlinkSync(filePath);
      console.log(`DLQ reprocessed: ${item.task_id}`);
    } catch (err) {
      // Leave in DLQ
    }
  }
}

// Retry DLQ every 5 minutes
setInterval(reprocessDeadLetters, 300000);

模式三：幂等处理——回调重复送达也不出错

CaptchaAI 可能因网络重试把同一个回调送达两次甚至更多次。处理程序必须是幂等的：

@app.route("/callback")
def idempotent_callback():
    task_id = request.args.get("id")
    solution = request.args.get("code")

    with lock:
        # Only process if not already handled
        if task_id in results:
            return "OK", 200  # Already processed — skip silently

        results[task_id] = solution
        pending_tasks.pop(task_id, None)

    return "OK", 200

三种模式怎么选

调用量不大，偶尔宕机——用回调 + 兜底轮询就够了。
调用量大，数据库可能扛不住——上死信队列。
可能有多个消费者处理同一条结果——做幂等处理。
生产环境、有 SLA 要求——三种模式一起上。

实践建议：不必一开始就三种模式全部上齐，先接兜底轮询，再按需叠加死信队列和幂等处理。

常见故障排查

兜底轮询又发现已交付的任务：回调和轮询器竞争，加一层幂等检查即可。

死信队列只进不出：重放程序没跑或本身报错，检查日志、确认底层问题已修复。

回调返回 200 但结果丢了：处理程序响应后才崩溃，落库放到响应之前，或用死信队列兜底。

兜底轮询请求量太大：陈旧任务堆积太多，调大超时阈值，查一下服务器可用性。

常见问题

回调地址必须是公网 HTTPS 吗？

建议是。CaptchaAI 需要能从公网访问到你的地址，HTTP 也能收，但生产环境推荐 HTTPS。没有公网域名时先用轮询兜底。

处理程序应该返回什么状态码？

始终返回 200。返回 4xx/5xx 没有帮助——CaptchaAI 不一定会重试。真正的失败交给死信队列或兜底轮询处理。

国内服务器收不到回调，怎么排查？

先确认防火墙放行了 CaptchaAI 的出站 IP 和端口，再查域名解析。基础设施没问题的话，多半是跨境网络抖动，打开兜底轮询即可。

轮询和回调同时开启，会重复扣费吗？

不会。CaptchaAI 按并发线程数计费，解决次数不设上限，轮询只是接口调用，不产生额外费用。

下一步

想要生产级的稳定回调处理？获取你的 CaptchaAI API 密钥，把这篇里的三种模式接进你的系统。

相关指南：

CaptchaAI 回调 URL 错误处理：重试和死信模式

回调为什么会悄悄丢结果

模式一：回调 + 兜底轮询双保险

Python

JavaScript

模式二：死信队列——失败的结果不能就地消失

Python

JavaScript

模式三：幂等处理——回调重复送达也不出错

三种模式怎么选

常见故障排查

常见问题

回调地址必须是公网 HTTPS 吗？

处理程序应该返回什么状态码？

国内服务器收不到回调，怎么排查？

轮询和回调同时开启，会重复扣费吗？

相关文章

下一步

CaptchaAI 回调 URL 设置：完整的 Webhook 指南

CaptchaAI Pingback 和任务通知模式

CaptchaAI Webhook 安全：验证回调签名

CAPTCHA 管道状态的 Discord Webhook 警报

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

用于 CaptchaAI Worker 部署的 Ansible Playbook