CAPTCHA Worker 健康检查：Liveness 与 Readiness 实践

凌晨三点，钉钉告警响了：worker 的 Pod 是 Running，但 15 分钟没解出一个任务——API Key 可能耗尽，也可能卡在死循环。Kubernetes 只看得到进程还在，看不出这些区别。给 worker 加上 liveness、readiness、dependency 三层探针，编排器才能分清“活着”和“能干活”，自动把流量切走。

健康检查的三个层次

Liveness（活跃度）：进程还响应吗？失败就重启容器。
Readiness（就绪度）：能接新任务吗？失败就暂停路由。
Dependency（依赖）：上游服务（如 CaptchaAI API）正常吗？失败就优雅降级。

运维阈值：先定义清楚“不健康”，再写探针代码

先把阈值定下来，否则健康检查要么形同虚设，要么疯狂抖动：

readiness 挡新任务，liveness 触发重启，余额告警捕捉吞吐量下降，三者不要合并成一个判断；同时要和队列深度、错误率挂钩，而不是只看进程存活时长。
阈值要让值班人员一眼判断“要不要处理”，避免噪声报警淹没真正的故障。
部署在阿里云 ACK、腾讯云 TKE 上，双 11、618 流量翻倍时，给 readiness 留启动宽限期（initialDelaySeconds），避免新 Pod 没预热完就被判“不健康”。

Python 实现：Flask 三层健康检查路由

下面这段 Flask 代码维护三层探针共用的健康状态，暴露 /health/live、/health/ready、/health/dependencies 三个端点，分别对应上面的 liveness、readiness、dependency，余额查询做了 60 秒缓存。

import requests
import time
import threading
from flask import Flask, jsonify
from dataclasses import dataclass, field

API_KEY = "YOUR_API_KEY"
RESULT_URL = "https://ocr.captchaai.com/res.php"

app = Flask(__name__)


@dataclass
class WorkerHealth:
    """Tracks worker health metrics."""
    started_at: float = field(default_factory=time.monotonic)
    last_solve_at: float = 0.0
    total_solved: int = 0
    total_failed: int = 0
    consecutive_failures: int = 0
    balance: float | None = None
    balance_checked_at: float = 0.0
    _lock: threading.Lock = field(default_factory=threading.Lock)

    def record_success(self):
        with self._lock:
            self.total_solved += 1
            self.last_solve_at = time.monotonic()
            self.consecutive_failures = 0

    def record_failure(self):
        with self._lock:
            self.total_failed += 1
            self.consecutive_failures += 1

    @property
    def success_rate(self) -> float:
        total = self.total_solved + self.total_failed
        return self.total_solved / total if total > 0 else 1.0

    @property
    def seconds_since_last_solve(self) -> float:
        if self.last_solve_at == 0:
            return time.monotonic() - self.started_at
        return time.monotonic() - self.last_solve_at


health = WorkerHealth()

# Thresholds
MAX_CONSECUTIVE_FAILURES = 10
MAX_SECONDS_WITHOUT_SOLVE = 600  # 10 minutes
MIN_BALANCE = 1.0


def check_balance() -> float | None:
    """Check CaptchaAI balance."""
    now = time.monotonic()
    # Cache balance for 60 seconds
    if health.balance is not None and now - health.balance_checked_at < 60:
        return health.balance

    try:
        resp = requests.get(RESULT_URL, params={
            "key": API_KEY, "action": "getbalance", "json": 1,
        }, timeout=10).json()
        health.balance = float(resp.get("request", 0))
        health.balance_checked_at = now
        return health.balance
    except Exception:
        return health.balance  # Return cached value on error


@app.route("/health/live")
def liveness():
    """Liveness probe — is the process responsive?"""
    return jsonify({"status": "ok", "uptime_s": int(time.monotonic() - health.started_at)}), 200


@app.route("/health/ready")
def readiness():
    """Readiness probe — can the worker accept tasks?"""
    issues = []

    # Check consecutive failures
    if health.consecutive_failures >= MAX_CONSECUTIVE_FAILURES:
        issues.append(f"consecutive_failures={health.consecutive_failures}")

    # Check time since last solve
    if health.total_solved > 0 and health.seconds_since_last_solve > MAX_SECONDS_WITHOUT_SOLVE:
        issues.append(f"no_solve_for={int(health.seconds_since_last_solve)}s")

    # Check balance
    balance = check_balance()
    if balance is not None and balance < MIN_BALANCE:
        issues.append(f"low_balance=${balance:.2f}")

    if issues:
        return jsonify({
            "status": "not_ready",
            "issues": issues,
            "stats": {
                "solved": health.total_solved,
                "failed": health.total_failed,
                "success_rate": round(health.success_rate, 3),
            },
        }), 503

    return jsonify({
        "status": "ready",
        "stats": {
            "solved": health.total_solved,
            "failed": health.total_failed,
            "success_rate": round(health.success_rate, 3),
            "balance": balance,
        },
    }), 200


@app.route("/health/dependencies")
def dependencies():
    """Check upstream dependencies."""
    checks = {}

    # CaptchaAI API reachability
    try:
        resp = requests.get(RESULT_URL, params={
            "key": API_KEY, "action": "getbalance", "json": 1,
        }, timeout=10)
        checks["captchaai_api"] = {
            "status": "ok" if resp.status_code == 200 else "degraded",
            "response_ms": int(resp.elapsed.total_seconds() * 1000),
        }
    except Exception as e:
        checks["captchaai_api"] = {"status": "down", "error": str(e)}

    all_ok = all(c["status"] == "ok" for c in checks.values())
    return jsonify({
        "status": "ok" if all_ok else "degraded",
        "checks": checks,
    }), 200 if all_ok else 503


# --- Worker loop (runs in background) ---

def worker_loop():
    """Simulated CAPTCHA solving worker."""
    while True:
        try:
            # ... solve CAPTCHA logic ...
            health.record_success()
        except Exception:
            health.record_failure()
        time.sleep(1)


threading.Thread(target=worker_loop, daemon=True).start()

Node.js 实现：Express 健康检查路由

Express 版本逻辑与 Flask 版本对应，同样缓存 60 秒余额，避免探针拖慢响应。

const express = require("express");

const API_KEY = "YOUR_API_KEY";
const RESULT_URL = "https://ocr.captchaai.com/res.php";

const app = express();

const health = {
  startedAt: Date.now(),
  lastSolveAt: 0,
  totalSolved: 0,
  totalFailed: 0,
  consecutiveFailures: 0,
  balance: null,
  balanceCheckedAt: 0,

  recordSuccess() {
    this.totalSolved++;
    this.lastSolveAt = Date.now();
    this.consecutiveFailures = 0;
  },

  recordFailure() {
    this.totalFailed++;
    this.consecutiveFailures++;
  },

  get successRate() {
    const total = this.totalSolved + this.totalFailed;
    return total > 0 ? this.totalSolved / total : 1;
  },
};

async function checkBalance() {
  if (health.balance !== null && Date.now() - health.balanceCheckedAt < 60000) {
    return health.balance;
  }
  try {
    const url = `${RESULT_URL}?key=${API_KEY}&action=getbalance&json=1`;
    const resp = await (await fetch(url)).json();
    health.balance = parseFloat(resp.request);
    health.balanceCheckedAt = Date.now();
    return health.balance;
  } catch {
    return health.balance;
  }
}

app.get("/health/live", (req, res) => {
  res.json({ status: "ok", uptimeMs: Date.now() - health.startedAt });
});

app.get("/health/ready", async (req, res) => {
  const issues = [];

  if (health.consecutiveFailures >= 10) {
    issues.push(`consecutive_failures=${health.consecutiveFailures}`);
  }

  if (health.totalSolved > 0) {
    const silentMs = Date.now() - health.lastSolveAt;
    if (silentMs > 600_000) {
      issues.push(`no_solve_for=${Math.round(silentMs / 1000)}s`);
    }
  }

  const balance = await checkBalance();
  if (balance !== null && balance < 1.0) {
    issues.push(`low_balance=$${balance.toFixed(2)}`);
  }

  const stats = {
    solved: health.totalSolved,
    failed: health.totalFailed,
    successRate: Math.round(health.successRate * 1000) / 1000,
    balance,
  };

  if (issues.length > 0) {
    return res.status(503).json({ status: "not_ready", issues, stats });
  }
  res.json({ status: "ready", stats });
});

app.get("/health/dependencies", async (req, res) => {
  const checks = {};
  try {
    const start = Date.now();
    const url = `${RESULT_URL}?key=${API_KEY}&action=getbalance&json=1`;
    const resp = await fetch(url);
    checks.captchaaiApi = {
      status: resp.ok ? "ok" : "degraded",
      responseMs: Date.now() - start,
    };
  } catch (e) {
    checks.captchaaiApi = { status: "down", error: e.message };
  }

  const allOk = Object.values(checks).every((c) => c.status === "ok");
  res.status(allOk ? 200 : 503).json({
    status: allOk ? "ok" : "degraded",
    checks,
  });
});

app.listen(8080, () => console.log("Health server on :8080"));

Kubernetes 探针配置

把上面两套端点接进 livenessProbe 和 readinessProbe：initialDelaySeconds 给新 Pod 留出启动和预热时间，failureThreshold 决定连续失败几次才真正判定异常。

apiVersion: apps/v1
kind: Deployment
metadata:
  name: captcha-worker
spec:
  replicas: 3
  template:
    spec:
      containers:

        - name: worker
          image: captcha-worker:latest
          ports:

            - containerPort: 8080
          livenessProbe:
            httpGet:
              path: /health/live
              port: 8080
            initialDelaySeconds: 10
            periodSeconds: 15
            failureThreshold: 3
          readinessProbe:
            httpGet:
              path: /health/ready
              port: 8080
            initialDelaySeconds: 5
            periodSeconds: 10
            failureThreshold: 2

响应码速查

/health/live：200 = 进程正常响应；503 = 进程冻结——需要重启
/health/ready：200 = 可以接受新任务；503 = 暂停派发任务
/health/dependencies：200 = 所有依赖都 OK；503 = 上游服务降级

常见故障排查

现象	原因	处理方式
worker 不断重启	liveness 阈值太严格	调大 `failureThreshold` 或 `periodSeconds`
启动阶段 worker 被判未就绪	未解出首个任务就判定超时	解出首个任务后再检查 `seconds_since_last_solve`
余额检查拖慢端点	每次请求都实时调用 API	给余额加 TTL 缓存（建议 60 秒）
端点自己崩溃	检查逻辑有未捕获异常	每个检查包一层 try/except，出错返回降级而非 500
依赖检查误报	余额检查时网络抖动	用失效才重新校验的缓存值，而非每次都请求

常见问题

Kubernetes 探针应该多久检查一次？

Liveness 每 10–30 秒一次，连续失败 3 次重启；readiness 每 5–10 秒一次，连续失败 2 次摘流量。探测越频繁，发现越快，开销也越大。

健康检查端点要不要调用 CaptchaAI API？

只在 readiness 和 dependency 检查里调用，且要缓存结果。liveness 探针不发外部请求，必须立刻响应证明进程还活着。

健康检查失败要不要立刻发告警？

建议分级：liveness 失败直接重启，无需人工介入；readiness 或依赖降级超过几分钟，再推给钉钉/企业微信值班通道，避免一失败就轰炸值班人员。

Liveness 和 Readiness 能不能共用同一个端点？

不建议。liveness 只答“进程还响应吗”，不该被依赖拖累；readiness 要综合余额、失败次数等状态。合并成一个端点，依赖抖动时容易误重启 Pod。

CaptchaAI 健康检查监控脚本

下一步

想让 worker 真正生产就绪？获取 CaptchaAI API 密钥，接上这篇教程里的三层探针。

CAPTCHA Worker 健康检查端点怎么写

健康检查的三个层次

运维阈值：先定义清楚“不健康”，再写探针代码

Python 实现：Flask 三层健康检查路由

Node.js 实现：Express 健康检查路由

Kubernetes 探针配置

响应码速查

常见故障排查

常见问题

Kubernetes 探针应该多久检查一次？

健康检查端点要不要调用 CaptchaAI API？

健康检查失败要不要立刻发告警？

Liveness 和 Readiness 能不能共用同一个端点？

相关文章

下一步

相关指南

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

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

用于 CaptchaAI Worker 部署的 Ansible Playbook

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

Azure Functions + CaptchaAI：云集成

使用 CaptchaAI 在 Python 中构建验证码求解队列