分布式 Worker 验证码会话状态管理：Redis 实战方案

多个 worker 一起识别同一个站点验证码时，为什么成功率反而会掉？通常是因为每个 worker 都在用自己的一套会话，目标站点看到不同 cookie、不同 IP、不同浏览器特征。把 cookie、token、代理绑定集中存进 Redis，让所有 worker 共享统一上下文，站点看到的才是连贯会话。

不共享会话会出什么问题

一个典型的分布式抓取集群，如果 worker 之间互不通气，会是这样：

Worker 1 → Login → Solve CAPTCHA → Get cookie A → Submit form ✅
Worker 2 → New session → Solve CAPTCHA → Get cookie B → Submit form ✅
Worker 3 → Reuse cookie A? → Cookie expired → Solve CAPTCHA → Fail ❌

Worker 3 想复用 Worker 1 的 cookie，却不知道已过期，白白浪费一次识别，还留下可被检测的不一致行为。

整体架构设计

┌──────────────────────────────────────┐
│          Session State Store          │
│              (Redis)                  │
│                                      │
│  cookies:{domain} → Hash             │
│  tokens:{sitekey} → List             │
│  proxies:pool → Set                  │
│  locks:{domain}:{worker} → String    │
└─────┬──────────┬──────────┬──────────┘
      │          │          │
  ┌───▼───┐  ┌──▼────┐  ┌──▼────┐
  │Worker1│  │Worker2│  │Worker3│
  └───────┘  └───────┘  └───────┘

真实场景： 阿里云 Kubernetes 集群跑 5 个抓取 Pod 识别 Turnstile；接入 Redis 后，新扩容的 Pod 几秒内即可复用其他 worker 的 cookie 和空闲 token。

哪些状态需要共享

不是所有状态都要共享，CSRF token 就必须各留一份：

状态组件	生命周期	共享策略
身份验证 cookie	几分钟到几小时	Redis + TTL
CAPTCHA token	90–300 秒	Redis 列表（短 TTL）
`qa_session_cookie` cookie	~30 分钟	Redis 哈希
CSRF token	每次页面加载	不共享——每个 worker 各自持有
浏览器特征	长期不变	写入配置，而非运行时状态
代理分配	每次会话	Redis 支持的代理池

Python 实现

会话存储类

import os
import json
import time
import redis
import requests
from datetime import datetime, timezone

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"]


class SessionStore:
    """Shared session state across distributed workers."""

    def __init__(self, domain):
        self.domain = domain
        self.cookie_key = f"session:cookies:{domain}"
        self.token_key = f"session:tokens:{domain}"

    def save_cookies(self, cookies, ttl=1800):
        """Store cookies from a successful session."""
        cookie_data = {name: value for name, value in cookies.items()}
        r.hset(self.cookie_key, mapping=cookie_data)
        r.expire(self.cookie_key, ttl)

    def get_cookies(self):
        """Retrieve shared cookies."""
        cookies = r.hgetall(self.cookie_key)
        return cookies if cookies else None

    def save_token(self, sitekey, token, ttl=80):
        """Store a solved CAPTCHA token."""
        key = f"{self.token_key}:{sitekey}"
        r.rpush(key, token)
        r.expire(key, ttl)

    def get_token(self, sitekey):
        """Pop a cached CAPTCHA token."""
        key = f"{self.token_key}:{sitekey}"
        return r.lpop(key)

    def acquire_session_lock(self, worker_id, ttl=300):
        """Ensure only one worker manages the session at a time."""
        lock_key = f"session:lock:{self.domain}"
        return r.set(lock_key, worker_id, nx=True, ex=ttl)

    def release_session_lock(self, worker_id):
        """Release session lock if this worker holds it."""
        lock_key = f"session:lock:{self.domain}"
        current = r.get(lock_key)
        if current == worker_id:
            r.delete(lock_key)

使用共享状态的 worker

worker 启动时先从 Redis 拉共享 cookie，识别前先查一次 token 缓存，命中就直接复用：

class CaptchaWorker:
    def __init__(self, worker_id, domain):
        self.worker_id = worker_id
        self.store = SessionStore(domain)
        self.session = requests.Session()

    def setup_session(self):
        """Load shared cookies into this worker's session."""
        cookies = self.store.get_cookies()
        if cookies:
            for name, value in cookies.items():
                self.session.cookies.set(name, value)
            return True
        return False

    def solve_captcha(self, sitekey, pageurl):
        """Solve with token cache and session sharing."""
        # Check for cached token
        cached = self.store.get_token(sitekey)
        if cached:
            return {"solution": cached, "source": "cache"}

        # Solve via 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:
            return {"error": data.get("request")}

        captcha_id = data["request"]

        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:
                token = result["request"]
                self.store.save_token(sitekey, token)
                return {"solution": token, "source": "api"}

            if result.get("request") != "CAPCHA_NOT_READY":
                return {"error": result.get("request")}

        return {"error": "TIMEOUT"}

    def process_page(self, url, sitekey):
        """Full workflow: setup session → solve CAPTCHA → submit."""
        # Load shared session
        self.setup_session()

        # Solve CAPTCHA
        result = self.solve_captcha(sitekey, url)
        if "error" in result:
            return result

        # Submit form with token
        response = self.session.post(url, data={
            "g-recaptcha-response": result["solution"]
        })

        # Share resulting cookies
        self.store.save_cookies(dict(self.session.cookies))

        return {"status": response.status_code, "source": result["source"]}

代理池管理

worker 一多容易忽视代理复用：两个 worker 共用代理登录同一域名，一眼就会被识破。用 Set 维护空闲代理池，用完归还：

class ProxyPool:
    """Distribute proxies across workers to avoid IP conflicts."""

    def __init__(self, proxies):
        self.pool_key = "session:proxy_pool"
        self.assigned_key = "session:proxy_assigned"
        # Initialize pool
        for proxy in proxies:
            r.sadd(self.pool_key, proxy)

    def acquire_proxy(self, worker_id, ttl=600):
        """Assign an unused proxy to a worker."""
        # Check if worker already has one
        existing = r.hget(self.assigned_key, worker_id)
        if existing:
            return existing

        # Pop from available pool
        proxy = r.spop(self.pool_key)
        if proxy:
            r.hset(self.assigned_key, worker_id, proxy)
            r.expire(self.assigned_key, ttl)
            return proxy
        return None

    def release_proxy(self, worker_id):
        """Return proxy to the pool."""
        proxy = r.hget(self.assigned_key, worker_id)
        if proxy:
            r.sadd(self.pool_key, proxy)
            r.hdel(self.assigned_key, worker_id)

JavaScript 实现

Node.js 版思路相同，ioredis 替代 redis-py：

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

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

class SessionStore {
  constructor(domain) {
    this.domain = domain;
    this.cookieKey = `session:cookies:${domain}`;
    this.tokenKey = `session:tokens:${domain}`;
  }

  async saveCookies(cookies, ttl = 1800) {
    const entries = Object.entries(cookies).flat();
    if (entries.length > 0) {
      await redis.hset(this.cookieKey, ...entries);
      await redis.expire(this.cookieKey, ttl);
    }
  }

  async getCookies() {
    return await redis.hgetall(this.cookieKey);
  }

  async saveToken(sitekey, token, ttl = 80) {
    const key = `${this.tokenKey}:${sitekey}`;
    await redis.rpush(key, token);
    await redis.expire(key, ttl);
  }

  async getToken(sitekey) {
    return await redis.lpop(`${this.tokenKey}:${sitekey}`);
  }

  async acquireLock(workerId, ttl = 300) {
    const result = await redis.set(`session:lock:${this.domain}`, workerId, "NX", "EX", ttl);
    return result === "OK";
  }

  async releaseLock(workerId) {
    const current = await redis.get(`session:lock:${this.domain}`);
    if (current === workerId) await redis.del(`session:lock:${this.domain}`);
  }
}

async function workerSolve(store, sitekey, pageurl) {
  const cached = await store.getToken(sitekey);
  if (cached) return { solution: cached, source: "cache" };

  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) 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 store.saveToken(sitekey, poll.data.request);
      return { solution: poll.data.request, source: "api" };
    }
    if (poll.data.request !== "CAPCHA_NOT_READY") return { error: poll.data.request };
  }
  return { error: "TIMEOUT" };
}

常见状态管理模式

会话锁：一个 worker 登录，其他 worker 复用其 cookie。
Token 池：高并发场景下批量识别，再按需分发给 worker。
Cookie 共享：worker 之间需要复用同一套已登录的会话。
代理亲和性：目标站点按 IP 与会话的绑定关系做风控。

常见故障排查

各 worker 拿到的会话不一致（cookie 没有通过 Redis 共享）：确认请求成功后调用了 save_cookies。
Token 还没被别的 worker 用就过期了（TTL 太长或网络延迟叠加）：缩小 TTL 余量，拿到后 10 秒内尽快使用。
会话锁一直没释放（持有锁的 worker 异常退出）：锁键自带 TTL（默认 300 秒），到期自动释放。
目标站点直接封了 worker（所有 worker 共用同一个代理出口）：换成带 worker 亲和性的代理池。

常见问题

只有站点要求已登录会话时才需要；无状态识别（提交 sitekey → 拿 token）只需共享 token。

按实际生命周期留安全余量：

CAPTCHA token：通常 90–300 秒失效，TTL 设 60–80 秒最稳
身份验证 cookie：TTL 建议比真实寿命略短，让 Redis 主动清理

Redis 挂了，所有 worker 会一起失败吗？

会话状态层有单点风险：

生产建议用 Redis Cluster 或哨兵模式
worker 拿不到共享 cookie 时退回独立登录，避免集群一起失败

基于浏览器的会话（Puppeteer/Playwright）怎么共享？

用 page.cookies() 把浏览器 cookie 存进 Redis，其他 worker 用 page.setCookie() 加载回来，思路与前面的 requests.Session 一致。

下一步

接入 worker 集群，剩下的识别交给 CaptchaAI：获取 API Key。

相关指南：

跨分布式工作人员的 CAPTCHA 会话状态管理

不共享会话会出什么问题

整体架构设计

哪些状态需要共享

Python 实现

会话存储类

使用共享状态的 worker

代理池管理

JavaScript 实现

常见状态管理模式

常见故障排查

常见问题

Redis 挂了，所有 worker 会一起失败吗？

基于浏览器的会话（Puppeteer/Playwright）怎么共享？

下一步

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

用于 CaptchaAI Worker 部署的 Ansible Playbook

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

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

Azure Functions + CaptchaAI：云集成

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

不共享会话会出什么问题

整体架构设计

哪些状态需要共享

Python 实现

会话存储类

使用共享状态的 worker

代理池管理

JavaScript 实现

常见状态管理模式

常见故障排查

常见问题

是不是所有 worker 都要共享 cookie？

Token 和 cookie 的 TTL 应该设多长？

Redis 挂了，所有 worker 会一起失败吗？

基于浏览器的会话（Puppeteer/Playwright）怎么共享？

下一步

相关文章

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

用于 CaptchaAI Worker 部署的 Ansible Playbook

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

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

Azure Functions + CaptchaAI：云集成

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