服务器发送的实时 CAPTCHA 解决通知事件 (实用指南)

每 5 秒轮询一次 res.php 是可行的，但它会浪费请求并增加延迟。服务器发送事件 (SSE) 让您的服务器可以在连接的客户端到达时立即向它们推送验证码解决方案 - 零浪费请求，亚秒级交付。

SSE 如何适应验证码工作流程

[Client] ← SSE stream ← [Your Server] ← Callback ← [CaptchaAI]
   ↓                          ↑
   Submit task → [CaptchaAI] ──┘ (pingback URL points to your server)

客户端连接到您的 SSE 端点（持久 HTTP 连接）
客户端向 CaptchaAI 提交验证码任务，其中 pingback 指向您的服务器
CaptchaAI 求解并将结果发送到您的回调端点
您的服务器通过 SSE 流将结果推送到客户端

完整实现——Python (Flask)

服务器

import os
import queue
import threading
import requests
from flask import Flask, Response, request, jsonify

app = Flask(__name__)

API_KEY = os.environ["CAPTCHAAI_API_KEY"]

# Per-client event queues: client_id -> Queue
client_queues = {}
queues_lock = threading.Lock()


@app.route("/events/<client_id>")
def sse_stream(client_id):
    """SSE endpoint — clients connect here for real-time results."""
    q = queue.Queue()

    with queues_lock:
        client_queues[client_id] = q

    def generate():
        try:
            while True:
                # Block until a result arrives (timeout for keepalive)
                try:
                    data = q.get(timeout=30)
                    yield f"event: captcha-solved\ndata: {data}\n\n"
                except queue.Empty:
                    # Send keepalive comment to prevent connection timeout
                    yield ": keepalive\n\n"
        finally:
            with queues_lock:
                client_queues.pop(client_id, None)

    return Response(
        generate(),
        mimetype="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no"  # Disable nginx buffering
        }
    )


@app.route("/submit", methods=["POST"])
def submit_captcha():
    """Submit a CAPTCHA task with callback to this server."""
    data = request.json
    client_id = data["client_id"]
    sitekey = data["sitekey"]
    pageurl = data["pageurl"]

    callback_url = f"{request.host_url}callback?client_id={client_id}"

    resp = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "pingback": callback_url,
        "json": 1
    })
    result = resp.json()

    if result.get("status") == 1:
        return jsonify({"task_id": result["request"]})
    return jsonify({"error": result.get("request")}), 400


@app.route("/callback")
def captcha_callback():
    """Receive CaptchaAI callback and push to SSE stream."""
    client_id = request.args.get("client_id")
    task_id = request.args.get("id")
    solution = request.args.get("code")

    import json
    message = json.dumps({
        "task_id": task_id,
        "solution": solution
    })

    with queues_lock:
        q = client_queues.get(client_id)
        if q:
            q.put(message)

    return "OK", 200


if __name__ == "__main__":
    app.run(port=5000, threaded=True)

浏览器客户端

<!DOCTYPE html>
<html>
<body>
  <button onclick="submitCaptcha()">Solve CAPTCHA</button>
  <div id="results"></div>

  <script>
    const clientId = crypto.randomUUID();
    const resultsDiv = document.getElementById("results");

    // Connect SSE stream
    const eventSource = new EventSource(`/events/${clientId}`);

    eventSource.addEventListener("captcha-solved", (event) => {
      const data = JSON.parse(event.data);
      resultsDiv.innerHTML += `<p>Task ${data.task_id}: ${data.solution.substring(0, 30)}...</p>`;
    });

    eventSource.onerror = () => {
      console.log("SSE connection lost, reconnecting...");
    };

    async function submitCaptcha() {
      const response = await fetch("/submit", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({
          client_id: clientId,
          sitekey: "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
          pageurl: "https://example.com"
        })
      });
      const result = await response.json();
      resultsDiv.innerHTML += `<p>Submitted: ${result.task_id}</p>`;
    }
  </script>
</body>
</html>

完整实现 - JavaScript (Express)

服务器

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

const app = express();
app.use(express.json());

const API_KEY = process.env.CAPTCHAAI_API_KEY;
const BASE_URL = process.env.BASE_URL || "http://localhost:3000";

// Per-client SSE connections: clientId -> Response object
const clients = new Map();

// SSE endpoint
app.get("/events/:clientId", (req, res) => {
  const clientId = req.params.clientId;

  res.writeHead(200, {
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache",
    Connection: "keep-alive",
    "X-Accel-Buffering": "no",
  });

  clients.set(clientId, res);

  // Keepalive every 30 seconds
  const keepalive = setInterval(() => {
    res.write(": keepalive\n\n");
  }, 30000);

  req.on("close", () => {
    clearInterval(keepalive);
    clients.delete(clientId);
  });
});

// Submit CAPTCHA
app.post("/submit", async (req, res) => {
  const { client_id, sitekey, pageurl } = req.body;
  const callbackUrl = `${BASE_URL}/callback?client_id=${client_id}`;

  try {
    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) {
      return res.json({ task_id: resp.data.request });
    }
    res.status(400).json({ error: resp.data.request });
  } catch (err) {
    res.status(500).json({ error: err.message });
  }
});

// CaptchaAI callback → push to SSE
app.get("/callback", (req, res) => {
  const clientId = req.query.client_id;
  const taskId = req.query.id;
  const solution = req.query.code;

  const clientRes = clients.get(clientId);
  if (clientRes) {
    const data = JSON.stringify({ task_id: taskId, solution: solution });
    clientRes.write(`event: captcha-solved\ndata: ${data}\n\n`);
  }

  res.sendStatus(200);
});

app.listen(3000, () => console.log("SSE server running on :3000"));

SSE、WebSocket、轮询

特征	上证所	WebSocket	轮询
方向	服务器 → 客户端	双向	客户端 → 服务器
协议	HTTP/1.1+	WS/WSS	HTTP协议
自动重新连接	内置	手动的	N/A
浏览器支持	全部现代	全部现代	全部
复杂	低的	中等的	低的
浪费的请求	没有任何	没有任何	许多
最适合验证码结果	是的	矫枉过正	有效但浪费

SSE 是验证码结果的理想选择，因为数据仅在服务器到客户端之间流动。

生产注意事项

通过多个服务器实例进行扩展

SSE 连接是有状态的 - 如果您的服务器在负载均衡器后面有多个实例，则回调可能会命中与持有客户端 SSE 连接的实例不同的实例。

解决方案： 使用Redis Pub/Sub作为消息总线：

# Callback handler publishes to Redis
import redis
r = redis.Redis()
r.publish(f"captcha:{client_id}", json.dumps(message))

# SSE handler subscribes to Redis
pubsub = r.pubsub()
pubsub.subscribe(f"captcha:{client_id}")
for msg in pubsub.listen():
    if msg["type"] == "message":
        yield f"data: {msg['data'].decode()}\n\n"

连接限制

浏览器将 SSE 连接限制为每个域 6 个 (HTTP/1.1)。使用 HTTP/2 获得更高的限制，或通过每个客户端的单个 SSE 连接复用多个任务结果。

故障排除

问题	原因	处理方式
SSE 连接每 30 秒就会断开一次	Proxy/load 平衡器超时	发送 keepalive 评论；增加代理超时
结果未到达	回调命中不同的服务器实例	在回调和 SSE 处理程序之间添加 Redis Pub/Sub
浏览器在控制台中显示错误	CORS 标头缺失	将 `Access-Control-Allow-Origin` 标头添加到 SSE 端点
多次重新连接	服务器发送格式错误的 SSE	确保 `\n\n` 终止每个事件；验证数据格式

常问问题

SSE 是否在 Cloudflare 背后运行？

是的，但 Cloudflare 可能会缓冲响应。使用 X-Accel-Buffering: no 标头禁用响应缓冲，或使用 Cloudflare 的流模式。

一台服务器可以处理多少个并发 SSE 连接？

Node.js 可以轻松处理 10,000 多个并发 SSE 连接，因为每个连接都是轻量级的保持活动 HTTP 连接。具有线程的 Python 受到更多限制 - 使用异步框架（带有 asyncio 的 FastAPI）来实现高并发。

我应该对非浏览器客户端使用 SSE 吗？

对于 CLI 工具或后端服务，直接回调处理或基于队列的方法更简单。当将结果推送到基于浏览器的仪表板或 Web 应用程序时，SSE 最有用。

下一步

实时传输验证码解决方案 –”获取您的 CaptchaAI API 密钥并将 SSE 连接到您的回调管道。

相关指南：

服务器发送的实时 CAPTCHA 解决通知事件

SSE 如何适应验证码工作流程

完整实现——Python (Flask)

服务器

浏览器客户端

完整实现 - JavaScript (Express)

服务器

SSE、WebSocket、轮询

生产注意事项

通过多个服务器实例进行扩展

连接限制

故障排除

常问问题

SSE 是否在 Cloudflare 背后运行？

一台服务器可以处理多少个并发 SSE 连接？

我应该对非浏览器客户端使用 SSE 吗？

下一步

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

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

用于 CaptchaAI Worker 部署的 Ansible Playbook

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

Azure Functions + CaptchaAI：云集成

构建 CaptchaAI 使用仪表板和监控

SSE 如何适应验证码工作流程

完整实现——Python (Flask)

服务器

浏览器客户端

完整实现 - JavaScript (Express)

服务器

SSE、WebSocket、轮询

生产注意事项

通过多个服务器实例进行扩展

连接限制

故障排除

常问问题

SSE 是否在 Cloudflare 背后运行？

一台服务器可以处理多少个并发 SSE 连接？

我应该对非浏览器客户端使用 SSE 吗？

下一步

相关文章

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

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

用于 CaptchaAI Worker 部署的 Ansible Playbook

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

Azure Functions + CaptchaAI：云集成

构建 CaptchaAI 使用仪表板和监控