Xây dựng hàng đợi giải CAPTCHA bằng Python với CaptchaAI

Nếu code của bạn đang giải CAPTCHA theo kiểu tuần tự — gửi task, chờ kết quả, rồi mới gửi task tiếp theo — phần lớn thời gian chạy chỉ để chờ. Cách xử lý đúng khi cần giải hàng trăm CAPTCHA là tách bước gửi task khỏi bước polling kết quả: gửi toàn bộ task tới in.php gần như cùng lúc, để nhiều worker cùng polling res.php và lấy token về song song. Bài này đi thẳng vào bốn mẫu hàng đợi bạn có thể copy vào dự án Python ngay hôm nay: threading cơ bản, asyncio bất đồng bộ, producer-consumer cho crawler chạy liên tục, và hàng đợi ưu tiên khi một số CAPTCHA (như trang checkout) cần giải trước các CAPTCHA khác.

Đây là bài toán quen thuộc với team automation/QA tại các công ty outsourcing hoặc sản phẩm ở Việt Nam — chẳng hạn đội scraping theo dõi giá trên Shopee, Lazada, Tiki, hay đội QA kiểm thử luồng đăng ký/checkout song song trên nhiều trang.

Vì sao giải tuần tự không mở rộng được?

Giải từng CAPTCHA một, chờ xong mới gửi cái tiếp theo, lãng phí phần lớn thời gian vào việc chờ mạng. Một hệ thống hàng đợi đúng cách sẽ:

• Gửi toàn bộ CAPTCHA gần như đồng thời thay vì xếp hàng chờ từng cái
• Polling nhiều ID task song song thay vì polling tuần tự
• Tự động thử lại khi một task lỗi, không cần can thiệp thủ công
• Giới hạn số worker đồng thời để không vượt rate limit của API
• Cung cấp theo dõi tiến độ và gọi callback ngay khi có kết quả

Bốn mẫu triển khai dưới đây đi từ đơn giản (threading) đến phức hợp (hàng đợi ưu tiên), bạn chọn theo mức độ phức tạp của workload thực tế.

Mẫu 1: hàng đợi threading cơ bản

Đây là cách nhanh nhất để thêm xử lý song song vào code Python đồng bộ hiện có, không cần viết lại toàn bộ ứng dụng theo asyncio. Một pool worker cố định liên tục lấy task từ Queue, gửi tới CaptchaAI, polling kết quả rồi đẩy vào result_queue:

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 là điểm khởi đầu an toàn cho gói BASIC ($15/tháng, 5 thread) — mỗi worker chiếm một thread trong lúc chờ polling, nên số worker không nên vượt số thread bạn đang trả tiền.

Mẫu 2: hàng đợi asyncio bất đồng bộ

Nếu dự án mới hoặc bạn đã dùng aiohttp/asyncio sẵn, mẫu này hiệu quả hơn threading vì giải CAPTCHA chủ yếu là chờ I/O (network), không tốn CPU. Semaphore giới hạn số task chạy đồng thời để không vượt giới hạn API:

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

return_exceptions=True trong asyncio.gather rất quan trọng: nếu bỏ nó, một task lỗi sẽ làm cả batch dừng thay vì chỉ báo lỗi task đó.

Mẫu 3: producer-consumer cho crawler chạy liên tục

Hai mẫu trên phù hợp khi biết trước danh sách CAPTCHA cần giải. Với crawler chạy liên tục — ví dụ theo dõi giá trên nhiều danh mục Shopee/Lazada, nơi trang mới liên tục xuất hiện trong lúc crawl — cần một producer đẩy task vào hàng đợi trong khi các consumer giải song song, không chờ toàn bộ danh sách có sẵn trước:

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

queue_size=100 giới hạn bộ nhớ đệm: nếu producer phát hiện trang mới nhanh hơn consumer giải xong, await self.queue.put(task) sẽ tự chờ thay vì để hàng đợi phình vô hạn.

Mẫu 4: hàng đợi ưu tiên khi một số CAPTCHA cần giải trước

Không phải CAPTCHA nào cũng quan trọng như nhau. Trong luồng kiểm thử checkout trên staging, CAPTCHA ở bước thanh toán thường cần token trước CAPTCHA ở các trang thông tin ít khẩn cấp hơn. asyncio.PriorityQueue xử lý đúng thứ tự này — số ưu tiên càng nhỏ càng được giải trước:

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

Đo throughput và chọn gói CaptchaAI phù hợp

Trước khi tăng max_workers hay num_consumers, hãy đo throughput thực tế thay vì đoán. QueueMetrics bên dưới tính thời gian giải trung bình, tỷ lệ giải thành công và số CAPTCHA giải được mỗi phút:

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

Nếu report() cho thấy throughput đứng yên dù đã tăng num_workers, nghẽn cổ chai thường nằm ở số thread của gói, không phải ở code. Số worker hợp lý là số nhỏ hơn hoặc bằng số thread của gói: BASIC ($15/tháng, 5 thread) hợp cho test; ADVANCE ($90/tháng, 50 thread) phù hợp hơn cho crawler chạy liên tục ở quy mô agency.

Khắc phục sự cố hàng đợi

Câu hỏi thường gặp

Nên chạy bao nhiêu worker đồng thời trong hàng đợi CAPTCHA?

Bắt đầu với 5–10 worker và tăng dần theo số thread của gói CaptchaAI bạn đang dùng. ERROR_NO_SLOT_AVAILABLE xuất hiện thường xuyên là dấu hiệu cần giảm worker lại, không phải tăng thêm.

Threading hay asyncio phù hợp hơn cho hàng đợi CAPTCHA?

Dùng asyncio cho dự án mới — giải CAPTCHA chủ yếu là chờ I/O nên ít overhead hơn threading. Dùng threading nếu đang tích hợp vào code đồng bộ sẵn có.

Chạy hàng đợi 50 worker cần gói CaptchaAI nào?

ADVANCE ($90/tháng, 50 thread) khớp trực tiếp vì mỗi worker chiếm một thread khi đang polling. Chạy 50 worker trên gói BASIC (5 thread) sẽ liên tục dính ERROR_NO_SLOT_AVAILABLE.

Hàng đợi CAPTCHA có làm tăng chi phí so với giải từng cái không?

Không. CaptchaAI tính phí theo thread đang chạy, không theo số lần giải — hàng đợi chỉ giúp dùng hết công suất thread bạn đã trả tiền thay vì để nó rảnh trong lúc chờ tuần tự.

Tóm tắt

Hàng đợi giải CAPTCHA tách việc gửi task khỏi việc polling, cho phép giải song song thay vì tuần tự với CaptchaAI. Chọn threading nếu đang tích hợp vào code đồng bộ sẵn có, asyncio cho dự án Python mới, producer-consumer cho crawler chạy liên tục, và hàng đợi ưu tiên khi CAPTCHA ở trang checkout cần giải trước các trang còn lại. Dù chọn mẫu nào, hãy đo throughput bằng QueueMetrics trước khi tăng worker.

Bài viết liên quan

• Hướng dẫn đầy đủ Playwright Python với CaptchaAI
• Xây dựng pipeline CAPTCHA cho client với CaptchaAI
• Xây dựng automation có trách nhiệm với CaptchaAI