验证码并发控制：信号量模式实战

并发数不是越大越好。CaptchaAI 按线程计费：套餐有多少线程，就允许多少个验证码同时在识别中。一次把 100 个任务全丢出去，超出的部分只换来 429 和重试。信号量就是在代码层面钉死这个上限——并发稳定在套餐线程数附近，不撞限流，也不让线程闲置。

先看一个真实场景：跨境比价采集的调参过程

一套跨境比价工具每天凌晨采集若干已获授权的海外站点，页面上多是 reCAPTCHA v2 或 Cloudflare Turnstile，国内自测环境里则常见 GeeTest（极验）滑块。调参过程很典型：

先用 STANDARD（$30/月，15 线程）跑通链路，信号量设 12。
凌晨批次跑了 4 个多小时，升级到 ADVANCE（$90/月，50 线程）。
信号量提到 40，同一批任务压到 1 小时出头。

一个坑：机器在国内时 reCAPTCHA 要加载 Google 域名下的脚本，网络不可达的超时容易被算成识别失败，要单独计数。采集范围只限自有或已授权站点，遵守 robots 协议与相关法规。

上限设成多少：从套餐线程数倒推

信号量就是个计数器：拿到许可减一，释放加一，减到 0 就排队。要定的只有初值，锚点很明确——套餐线程数。

套餐	线程数	建议信号量初值
BASIC（$15/月）	5	5
STANDARD（$30/月）	15	12–15
ADVANCE（$90/月）	50	40–50
PREMIUM（$170/月）	100	80–100

更高的 CORPORATE、ENTERPRISE 与 VIP 系列同理。每线程识别次数不限，吞吐量只取决于线程数和识别耗时，留 10%–20% 余量给重试通常比顶满更快。动手前先确认 API Key 就绪、依赖装好（国内可用 pip install -i 走清华 TUNA 等镜像）。

Python 实现：asyncio.Semaphore

import asyncio
import aiohttp

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


async def solve_one(session, sem, sitekey, page_url):
    """Solve one CAPTCHA within the semaphore limit."""
    async with sem:
        # Submit
        async with session.post(SUBMIT_URL, data={
            "key": API_KEY,
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": page_url,
            "json": "1",
        }) as resp:
            data = await resp.json()

        if data["status"] != 1:
            return {"url": page_url, "error": data["request"]}

        task_id = data["request"]

        # Poll (still within semaphore)
        for _ in range(24):
            await asyncio.sleep(5)
            async with session.get(RESULT_URL, params={
                "key": API_KEY, "action": "get", "id": task_id, "json": "1"
            }) as resp:
                result = await resp.json()

            if result["status"] == 1:
                return {"url": page_url, "token": result["request"]}
            if result["request"] != "CAPCHA_NOT_READY":
                return {"url": page_url, "error": result["request"]}

        return {"url": page_url, "error": "TIMEOUT"}


async def solve_batch(tasks, max_concurrent=20):
    sem = asyncio.Semaphore(max_concurrent)

    async with aiohttp.ClientSession() as session:
        coros = [
            solve_one(session, sem, t["sitekey"], t["url"])
            for t in tasks
        ]
        results = await asyncio.gather(*coros)

    solved = sum(1 for r in results if "token" in r)
    print(f"Solved {solved}/{len(results)}")
    return results

这一版有个浪费：in.php 通常 1 秒内返回，res.php 轮询要等十几到几十秒，慢的那一半长期占着许可。拆成两个信号量后流转更快：

async def solve_split_sems(session, submit_sem, poll_sem, sitekey, page_url):
    # Submit phase — short, limited to 30 concurrent
    async with submit_sem:
        async with session.post(SUBMIT_URL, data={
            "key": API_KEY,
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": page_url,
            "json": "1",
        }) as resp:
            data = await resp.json()

    if data["status"] != 1:
        return {"error": data["request"]}

    task_id = data["request"]

    # Poll phase — longer, limited to 50 concurrent
    async with poll_sem:
        for _ in range(24):
            await asyncio.sleep(5)
            async with session.get(RESULT_URL, params={
                "key": API_KEY, "action": "get", "id": task_id, "json": "1"
            }) as resp:
                result = await resp.json()

            if result["status"] == 1:
                return {"token": result["request"]}
            if result["request"] != "CAPCHA_NOT_READY":
                return {"error": result["request"]}

    return {"error": "TIMEOUT"}


async def main(tasks):
    submit_sem = asyncio.Semaphore(30)  # 30 concurrent submits
    poll_sem = asyncio.Semaphore(50)     # 50 concurrent polls

    async with aiohttp.ClientSession() as session:
        coros = [
            solve_split_sems(session, submit_sem, poll_sem, t["sitekey"], t["url"])
            for t in tasks
        ]
        return await asyncio.gather(*coros)

Node.js 实现：手写一个信号量类

Node.js 没有内置信号量，用 Promise 队列实现只要二十来行：许可未满直接放行，满了就把 resolve 压入队列等待。

class Semaphore {
  constructor(max) {
    this.max = max;
    this.current = 0;
    this.queue = [];
  }

  acquire() {
    return new Promise(resolve => {
      if (this.current < this.max) {
        this.current++;
        resolve();
      } else {
        this.queue.push(resolve);
      }
    });
  }

  release() {
    this.current--;
    if (this.queue.length > 0) {
      this.current++;
      const next = this.queue.shift();
      next();
    }
  }
}

接进识别流程时关键在 try/finally：无论成功、报错还是超时，release() 都必须执行，否则整批卡死。

const axios = require('axios');

const API_KEY = 'YOUR_API_KEY';
const sem = new Semaphore(20);

async function solveOne(sitekey, pageUrl) {
  await sem.acquire();
  try {
    const submit = await axios.post('https://ocr.captchaai.com/in.php', null, {
      params: {
        key: API_KEY,
        method: 'userrecaptcha',
        googlekey: sitekey,
        pageurl: pageUrl,
        json: 1,
      },
    });

    if (submit.data.status !== 1) {
      return { url: pageUrl, error: submit.data.request };
    }

    const taskId = submit.data.request;

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

      if (poll.data.status === 1) {
        return { url: pageUrl, token: poll.data.request };
      }
      if (poll.data.request !== 'CAPCHA_NOT_READY') {
        return { url: pageUrl, error: poll.data.request };
      }
    }
    return { url: pageUrl, error: 'TIMEOUT' };
  } finally {
    sem.release();
  }
}

// Solve 100 tasks with max 20 concurrent
async function solveBatch(tasks) {
  const results = await Promise.all(
    tasks.map(t => solveOne(t.sitekey, t.url))
  );
  const solved = results.filter(r => r.token).length;
  console.log(`Solved: ${solved}/${results.length}`);
  return results;
}

让并发自己调节：自适应信号量

固定值适合稳定负载。混跑 reCAPTCHA v2、Turnstile、GeeTest v3 等耗时不同的类型时，可按窗口内错误率自动升降并发。

class AdaptiveSemaphore:
    def __init__(self, initial=20, min_val=5, max_val=50):
        self.value = initial
        self.min_val = min_val
        self.max_val = max_val
        self.sem = asyncio.Semaphore(initial)
        self.success_count = 0
        self.error_count = 0

    async def acquire(self):
        await self.sem.acquire()

    def release(self, success=True):
        self.sem.release()
        if success:
            self.success_count += 1
        else:
            self.error_count += 1

        total = self.success_count + self.error_count
        if total % 20 == 0:
            self._adjust()

    def _adjust(self):
        error_rate = self.error_count / (self.success_count + self.error_count)

        if error_rate > 0.2 and self.value > self.min_val:
            self.value = max(self.min_val, self.value - 5)
            self.sem = asyncio.Semaphore(self.value)
            print(f"Reduced concurrency to {self.value}")
        elif error_rate < 0.05 and self.value < self.max_val:
            self.value = min(self.max_val, self.value + 5)
            self.sem = asyncio.Semaphore(self.value)
            print(f"Increased concurrency to {self.value}")

        self.success_count = 0
        self.error_count = 0

上下边界必须设，否则一段网络抖动就能把并发压到 1；只把业务错误计入 error_count。

常见故障与排查

现象	根因	处理方式
任务全部卡住	异常路径没释放许可	用 `try/finally` 或 `async with`
仍频繁 429	初值高于套餐线程数	下调初值，留 10%–20% 余量
整批比预期慢	初值太低或提交被轮询挤占	提高初值，或拆分两个信号量
内存持续增长	任务无限期排队等许可	给 `acquire()` 加超时
成功率忽高忽低	网络超时被算进识别失败	超时与业务错误分开统计

常见问题

信号量的值应该设成多少？

按套餐线程数设：BASIC（$15/月）5 线程设 5，ADVANCE（$90/月）50 线程设 40–50。跑一段看 429 比例，没有就小幅上调，一出现就回退。

提交和轮询要分别限流吗？

值得分。共用一个许可时慢的轮询长期占着它，提交端排不上队。拆开后提交端收紧、轮询端放宽即可。

多台机器同时跑，进程内信号量还管用吗？

只约束自己。多机部署最简单的做法是每台设成「总线程数 ÷ 机器数」，负载不均时改用 Redis 计数器做跨进程许可。

并发调高了，识别成功率会下降吗？

识别成功率取决于验证码类型和参数是否正确，与并发数无关。并发过高影响的是请求是否被接受——被 429 拒掉的任务没进入识别环节，却会在统计里算作失败。请在自有环境中实测。

把并发控制交给信号量

在 captchaai.com 注册拿到 API Key，按套餐线程数设好初值，先跑 100 个任务确认没有 429，再上调。

CAPTCHA 并发控制的信号量模式

先看一个真实场景：跨境比价采集的调参过程

上限设成多少：从套餐线程数倒推

Python 实现：asyncio.Semaphore

Node.js 实现：手写一个信号量类

让并发自己调节：自适应信号量

常见故障与排查

常见问题

信号量的值应该设成多少？

提交和轮询要分别限流吗？

多台机器同时跑，进程内信号量还管用吗？

并发调高了，识别成功率会下降吗？

把并发控制交给信号量

延伸阅读

自动缩放验证码解决工作人员

验证码解决率下降：性能回归诊断

验证码解决吞吐量：如何每小时处理 10,000 个任务

验证码识别 Worker 横向扩容：什么时候扩、怎么扩

并行与顺序验证码解决：性能权衡

限制您自己的验证码解决请求的速率

先看一个真实场景：跨境比价采集的调参过程

上限设成多少：从套餐线程数倒推

Python 实现：asyncio.Semaphore

Node.js 实现：手写一个信号量类

让并发自己调节：自适应信号量

常见故障与排查

常见问题

信号量的值应该设成多少？

提交和轮询要分别限流吗？

多台机器同时跑，进程内信号量还管用吗？

并发调高了，识别成功率会下降吗？

把并发控制交给信号量

延伸阅读

相关文章

自动缩放验证码解决工作人员

验证码解决率下降：性能回归诊断

验证码解决吞吐量：如何每小时处理 10,000 个任务

验证码识别 Worker 横向扩容：什么时候扩、怎么扩

并行与顺序验证码解决：性能权衡

限制您自己的验证码解决请求的速率