CAPTCHA 并发控制的信号量模式

同时发送 100 个验证码任务听起来很快 - 直到达到速率限制。信号量可让您将并发 API 调用的数量限制在安全水平，同时保持高吞吐量。本指南介绍了用于控制 Python 和 Node.js 中 CaptchaAI 并发性的信号量模式。

为什么使用信号量

信号量是一个计数器，限制可以同时运行的任务数量。当任务获取信号量时，计数就会减少。完成后，计数会增加。如果计数为零，则新任务等待。

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

拆分提交和轮询信号量

提交速度很快（< 1 秒）。轮询需要 15-60 秒。使用单独的信号量可以让您更快地提交：

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 没有内置信号量，但构建起来很简单：

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();
    }
  }
}

使用信号量

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;
}

自适应信号量

根据错误率调整并发：

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

故障排除

常问问题

我应该使用什么并发级别？

从 20 个并发任务开始。如果没有看到 429 错误，请增加到 30-50。如果发现错误，请减少到 10-15。

我应该限制提交或投票吗？

限制更积极地提交。轮询是轻量级的 GET 请求，可以以更高的并发性运行。

使用 CaptchaAI 扩展您的验证码解决方案

获取您的 API 密钥：验证码网站。

相关指南

• Python asyncio并发求解
• 并行验证码求解
• 速率限制和限制