Python 验证码识别队列搭建：线程与 asyncio

验证码识别队列的关键只有一件事：把"提交"和"轮询"拆开。拆开之后，并发数就能从 1 提到几十甚至上百，不用等一个验证码解决完才提交下一个。这在大规模抓取场景下尤其重要——不管抓的是海外站点（常见 reCAPTCHA v2、Turnstile）还是国内站点（更常见 GeeTest 极验），队列这套模型是通用的。

本文给出四种实现——线程池、asyncio、生产者消费者、优先级调度——按项目规模选一种即可，代码可直接复制使用。

为什么需要队列，而不是逐个提交？

逐个提交、逐个等待，是验证码识别环节最容易被忽视的性能瓶颈。队列系统把这件事做对：

能力	说明
批量提交	一次性提交所有任务
并行轮询	同时轮询多个 task_id
自动重试	失败自动重新提交
并发控制	限制并发，避免触发限流
进度追踪	提供回调与进度反馈

方案一：基于线程的队列

最简单的实现方式：标准库 threading + queue.Queue，适合已有同步代码的项目，无需引入 asyncio。

import time
import threading
import requests
from queue import Queue, Empty

API_KEY = "YOUR_API_KEY"


class CaptchaQueue:
    """Thread-based CAPTCHA solving queue."""

    def __init__(self, api_key, max_workers=10):
        self.api_key = api_key
        self.task_queue = Queue()
        self.result_queue = Queue()
        self.max_workers = max_workers
        self.workers = []

    def submit(self, method, callback=None, **params):
        """Add a CAPTCHA task to the queue."""
        task = {
            "method": method,
            "params": params,
            "callback": callback,
        }
        self.task_queue.put(task)

    def start(self):
        """Start worker threads."""
        for _ in range(self.max_workers):
            t = threading.Thread(target=self._worker, daemon=True)
            t.start()
            self.workers.append(t)

    def wait(self):
        """Wait for all tasks to complete."""
        self.task_queue.join()

    def get_results(self):
        """Get all available results."""
        results = []
        while not self.result_queue.empty():
            try:
                results.append(self.result_queue.get_nowait())
            except Empty:
                break
        return results

    def _worker(self):
        while True:
            try:
                task = self.task_queue.get(timeout=1)
            except Empty:
                continue

            try:
                result = self._solve(task["method"], **task["params"])
                entry = {"status": "solved", "result": result, "task": task}
                self.result_queue.put(entry)
                if task["callback"]:
                    task["callback"](result)
            except Exception as e:
                entry = {"status": "error", "error": str(e), "task": task}
                self.result_queue.put(entry)
            finally:
                self.task_queue.task_done()

    def _solve(self, method, **params):
        submit = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key, "method": method, "json": 1, **params,
        }, timeout=30).json()

        if submit.get("status") != 1:
            raise Exception(f"Submit error: {submit.get('request')}")

        task_id = submit["request"]
        for _ in range(30):
            time.sleep(5)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get", "id": task_id, "json": 1,
            }, timeout=30).json()
            if result.get("status") == 1:
                return result["request"]
            if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
                raise Exception("CAPTCHA unsolvable")
        raise TimeoutError("Solve timed out")


# Usage
queue = CaptchaQueue(API_KEY, max_workers=5)
queue.start()

# Submit multiple CAPTCHAs
urls_and_sitekeys = [
    ("https://example.com/page1", "SITEKEY_1"),
    ("https://example.com/page2", "SITEKEY_2"),
    ("https://example.com/page3", "SITEKEY_3"),
]

for url, sitekey in urls_and_sitekeys:
    queue.submit("userrecaptcha", googlekey=sitekey, pageurl=url)

queue.wait()
results = queue.get_results()
print(f"Solved {len(results)} CAPTCHAs")
for r in results:
    print(f"  {r['status']}: {r.get('result', r.get('error', ''))[:50]}")

max_workers 建议从 5 起步，再根据账号线程配额往上调，遇到瓶颈对照下文故障排查表定位。

方案二：asyncio 异步队列

新项目更推荐 asyncio：I/O 密集型请求用协程比线程更省资源，单进程也能撑起更高并发。

import asyncio
import aiohttp

API_KEY = "YOUR_API_KEY"


class AsyncCaptchaQueue:
    """Async CAPTCHA solving queue with concurrency control."""

    def __init__(self, api_key, max_concurrent=10):
        self.api_key = api_key
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.results = []

    async def solve_batch(self, tasks):
        """Solve a batch of CAPTCHA tasks concurrently."""
        coros = [self._solve_task(task) for task in tasks]
        self.results = await asyncio.gather(*coros, return_exceptions=True)
        return self.results

    async def _solve_task(self, task):
        async with self.semaphore:
            return await self._solve(task["method"], **task["params"])

    async def _solve(self, method, **params):
        async with aiohttp.ClientSession() as session:
            # Submit
            async with session.post("https://ocr.captchaai.com/in.php", data={
                "key": self.api_key, "method": method, "json": 1, **params,
            }) as resp:
                data = await resp.json(content_type=None)
                if data.get("status") != 1:
                    raise Exception(f"Submit error: {data.get('request')}")
                task_id = data["request"]

            # Poll
            for _ in range(30):
                await asyncio.sleep(5)
                async with session.get("https://ocr.captchaai.com/res.php", params={
                    "key": self.api_key, "action": "get", "id": task_id, "json": 1,
                }) as resp:
                    result = await resp.json(content_type=None)
                    if result.get("status") == 1:
                        return result["request"]
                    if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
                        raise Exception("CAPTCHA unsolvable")

            raise TimeoutError("Solve timed out")


# Usage
async def main():
    queue = AsyncCaptchaQueue(API_KEY, max_concurrent=5)

    tasks = [
        {"method": "userrecaptcha", "params": {"googlekey": f"SITEKEY_{i}", "pageurl": f"https://example.com/page{i}"}}
        for i in range(10)
    ]

    results = await queue.solve_batch(tasks)
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Task {i}: ERROR — {result}")
        else:
            print(f"Task {i}: {result[:50]}...")


asyncio.run(main())

方案三：生产者消费者模式，应对持续抓取

如果页面是动态发现的——翻页时才知道下一批 URL——一次性提交就不合适，生产者消费者模式更适合这种持续输入的场景：

import asyncio
import aiohttp

API_KEY = "YOUR_API_KEY"


class ProducerConsumerQueue:
    """Continuous CAPTCHA solving with producer-consumer pattern."""

    def __init__(self, api_key, queue_size=100, num_consumers=5):
        self.api_key = api_key
        self.queue = asyncio.Queue(maxsize=queue_size)
        self.num_consumers = num_consumers
        self.solved_count = 0
        self.error_count = 0
        self.running = True

    async def produce(self, tasks):
        """Producer: feed CAPTCHA tasks into the queue."""
        for task in tasks:
            await self.queue.put(task)
        # Signal consumers to stop
        for _ in range(self.num_consumers):
            await self.queue.put(None)

    async def consume(self, result_handler):
        """Consumer: solve CAPTCHAs and call result handler."""
        async with aiohttp.ClientSession() as session:
            while True:
                task = await self.queue.get()
                if task is None:
                    self.queue.task_done()
                    break

                try:
                    result = await self._solve(session, task["method"], **task["params"])
                    self.solved_count += 1
                    if result_handler:
                        await result_handler(task, result)
                except Exception as e:
                    self.error_count += 1
                    print(f"Error: {e}")
                finally:
                    self.queue.task_done()

    async def run(self, tasks, result_handler=None):
        """Run the producer-consumer pipeline."""
        # Start producer
        producer = asyncio.create_task(self.produce(tasks))

        # Start consumers
        consumers = [
            asyncio.create_task(self.consume(result_handler))
            for _ in range(self.num_consumers)
        ]

        # Wait for everything to finish
        await producer
        await asyncio.gather(*consumers)

        print(f"Complete: {self.solved_count} solved, {self.error_count} errors")

    async def _solve(self, session, method, **params):
        async with session.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key, "method": method, "json": 1, **params,
        }) as resp:
            data = await resp.json(content_type=None)
            if data.get("status") != 1:
                raise Exception(f"Submit: {data.get('request')}")
            task_id = data["request"]

        for _ in range(30):
            await asyncio.sleep(5)
            async with session.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get", "id": task_id, "json": 1,
            }) as resp:
                result = await resp.json(content_type=None)
                if result.get("status") == 1:
                    return result["request"]
        raise TimeoutError("Timed out")


# Usage
async def handle_result(task, token):
    url = task["params"]["pageurl"]
    print(f"Solved for {url}: {token[:30]}...")


async def main():
    queue = ProducerConsumerQueue(API_KEY, num_consumers=5)

    tasks = [
        {"method": "userrecaptcha", "params": {"googlekey": f"SITEKEY_{i}", "pageurl": f"https://example.com/page{i}"}}
        for i in range(20)
    ]

    await queue.run(tasks, result_handler=handle_result)


asyncio.run(main())

方案四：按优先级调度任务

不是所有验证码都同等重要：结账页的 Cloudflare Turnstile 应比商品页的 reCAPTCHA 更早处理，晚一步可能导致用户弃单。优先队列可以实现这种调度：

import asyncio
from dataclasses import dataclass, field

API_KEY = "YOUR_API_KEY"


@dataclass(order=True)
class PriorityTask:
    priority: int
    task: dict = field(compare=False)


class PriorityCaptchaQueue:
    """CAPTCHA queue with priority levels."""

    def __init__(self, api_key, num_workers=5):
        self.api_key = api_key
        self.queue = asyncio.PriorityQueue()
        self.num_workers = num_workers
        self.results = {}

    async def submit(self, task_id, method, priority=5, **params):
        """Submit with priority (lower number = higher priority)."""
        await self.queue.put(PriorityTask(
            priority=priority,
            task={"id": task_id, "method": method, "params": params},
        ))

    async def process(self):
        """Process all queued tasks by priority."""
        workers = [asyncio.create_task(self._worker()) for _ in range(self.num_workers)]

        # Wait for queue to drain
        await self.queue.join()

        # Cancel workers
        for w in workers:
            w.cancel()

        return self.results

    async def _worker(self):
        import aiohttp
        async with aiohttp.ClientSession() as session:
            while True:
                item = await self.queue.get()
                task = item.task
                try:
                    result = await self._solve(session, task["method"], **task["params"])
                    self.results[task["id"]] = {"status": "solved", "token": result}
                except Exception as e:
                    self.results[task["id"]] = {"status": "error", "error": str(e)}
                finally:
                    self.queue.task_done()

    async def _solve(self, session, method, **params):
        import aiohttp
        async with session.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key, "method": method, "json": 1, **params,
        }) as resp:
            data = await resp.json(content_type=None)
            if data.get("status") != 1:
                raise Exception(data.get("request"))
            task_id = data["request"]

        for _ in range(30):
            await asyncio.sleep(5)
            async with session.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get", "id": task_id, "json": 1,
            }) as resp:
                result = await resp.json(content_type=None)
                if result.get("status") == 1:
                    return result["request"]
        raise TimeoutError()


# Usage
async def main():
    pq = PriorityCaptchaQueue(API_KEY, num_workers=3)

    # High priority — checkout pages
    await pq.submit("checkout_1", "turnstile", priority=1, sitekey="KEY", pageurl="https://shop.com/checkout")

    # Normal priority — product pages
    for i in range(5):
        await pq.submit(f"product_{i}", "userrecaptcha", priority=5, googlekey="KEY", pageurl=f"https://shop.com/p/{i}")

    # Low priority — info pages
    for i in range(3):
        await pq.submit(f"info_{i}", "userrecaptcha", priority=10, googlekey="KEY", pageurl=f"https://shop.com/info/{i}")

    results = await pq.process()
    for task_id, result in results.items():
        print(f"{task_id}: {result['status']}")


asyncio.run(main())

队列运行数据怎么看：监控与统计

上线前建议加一层统计，方便判断并发设置是否合理，排查是否撞到限流：

import time
from dataclasses import dataclass, field


@dataclass
class QueueMetrics:
    submitted: int = 0
    solved: int = 0
    failed: int = 0
    total_solve_time: float = 0.0
    start_time: float = field(default_factory=time.time)

    @property
    def avg_solve_time(self):
        return self.total_solve_time / self.solved if self.solved else 0

    @property
    def success_rate(self):
        total = self.solved + self.failed
        return (self.solved / total * 100) if total else 0

    @property
    def throughput(self):
        elapsed = time.time() - self.start_time
        return self.solved / elapsed * 60 if elapsed > 0 else 0

    def report(self):
        return (
            f"Submitted: {self.submitted} | "
            f"Solved: {self.solved} | "
            f"Failed: {self.failed} | "
            f"Avg time: {self.avg_solve_time:.1f}s | "
            f"Success: {self.success_rate:.1f}% | "
            f"Throughput: {self.throughput:.0f}/min"
        )

常见故障排查

症状	原因	处理方式
队列在涨，但任务迟迟不完成	worker 数量过多，把 API 压垮了	调低 `max_workers` / `max_concurrent`
提示 `ERROR_NO_SLOT_AVAILABLE`	触发了账号的并发线程上限	在提交之间加一点延迟，或升级套餐线程数
任务卡在队列里不动	worker 线程因异常而退出	把 worker 循环包在 try/except 里
内存随时间持续增长	结果没有被消费	定期调用 `get_results()` 清空结果队列
异步队列卡死	漏加了 `await`	检查所有异步调用是否都写了 `await`

常见问题

worker 数量 / 并发数该怎么定？

别只看 CPU 核数，先看账号线程配额：BASIC 5 线程到 VIP-3 5,000 线程不等。max_workers/max_concurrent 设为线程配额的 60%–80% 比较稳，出现 ERROR_NO_SLOT_AVAILABLE 再调低。

线程池和 asyncio 该选哪个？

新项目直接上 asyncio，处理 I/O 密集型请求更省资源；接入已有同步代码库则用线程池改动最小。核心逻辑一致，都是把"提交"和"轮询"拆开。

一直报 ERROR_NO_SLOT_AVAILABLE 怎么办？

通常是账号并发线程数打满了，不是代码 bug。先确认 max_workers 未超过套餐线程数，再排查是否有任务卡住未释放——常见原因是 worker 异常退出后没调用 task_done()。

队列跑着跑着卡住不返回结果，怎么排查？

先看 worker 是否因未捕获异常"悄悄死掉"——try/except 包住 worker 循环即可避免；再检查 _solve() 里每个异步调用是否都写了 await，漏一个协程就会永远卡住。

总结

验证码识别队列的本质：把提交和轮询拆成两个环节，从"一次一个"变成"批量并行"。同步项目用线程池，新项目优先 asyncio，持续抓取用生产者消费者模式，需要分优先级就上优先队列——搭配 CaptchaAI 的并发线程额度即可跑通。

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

为什么需要队列，而不是逐个提交？

方案一：基于线程的队列

方案二：asyncio 异步队列

方案三：生产者消费者模式，应对持续抓取

方案四：按优先级调度任务

队列运行数据怎么看：监控与统计

常见故障排查

常见问题

worker 数量 / 并发数该怎么定？

线程池和 asyncio 该选哪个？

一直报 ERROR_NO_SLOT_AVAILABLE 怎么办？

队列跑着跑着卡住不返回结果，怎么排查？

总结

相关阅读

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

用于 CaptchaAI Worker 部署的 Ansible Playbook

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

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

Azure Functions + CaptchaAI：云集成

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

为什么需要队列，而不是逐个提交？

方案一：基于线程的队列

方案二：asyncio 异步队列

方案三：生产者消费者模式，应对持续抓取

方案四：按优先级调度任务

队列运行数据怎么看：监控与统计

常见故障排查

常见问题

worker 数量 / 并发数该怎么定？

线程池和 asyncio 该选哪个？

一直报 ERROR_NO_SLOT_AVAILABLE 怎么办？

队列跑着跑着卡住不返回结果，怎么排查？

总结

相关阅读

相关文章

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

用于 CaptchaAI Worker 部署的 Ansible Playbook

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

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

Azure Functions + CaptchaAI：云集成

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