Если парсер отправляет CAPTCHA по одной и ждёт ответа перед следующим запросом, узкое место — не сайт-источник, а ваш собственный код. Очередь решает это: сотни задач уходят в API CaptchaAI параллельно, а результаты забираются по мере готовности. Ниже — четыре рабочие реализации на Python: от простой очереди на потоках до приоритетной очереди для чувствительных к порядку сценариев, плюс мониторинг и разбор типичных ошибок.
Типичный кейс: команда, которая парсит цены на маркетплейсах для мониторинга конкурентов, подключает тариф ADVANCE ($90/мес, 50 потоков) и выставляет max_workers=50 — ровно по числу оплаченных потоков, без простоя и без превышения лимита.
Зачем нужна очередь при решении CAPTCHA
Решение CAPTCHA по одному запросу означает, что весь пайплайн простаивает, пока ждёт ответа от res.php. Очередь убирает это ожидание из основного потока выполнения: все CAPTCHA уходят в API сразу, а не по одной, несколько ID задач опрашиваются параллельно, неудачные попытки решения повторяются автоматически, параллелизм ограничивается так, чтобы не превысить лимит тарифа, а прогресс отслеживается через callback по каждому результату.
Многопоточная очередь на threading
Когда выбрать этот вариант
Самый простой вариант — пул потоков поверх queue.Queue. Каждый воркер получает задачу, отправляет её в in.php, опрашивает res.php и кладёт результат в очередь ответов. Такой подход хорошо ложится на существующий синхронный код парсера — не нужно переписывать весь пайплайн под 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]}")
В примере выше пять воркеров разбирают три CAPTCHA параллельно: queue.wait() блокирует основной поток выполнения до task_done() по каждой задаче, а get_results() забирает всё, что успело решиться. Если воркеров больше, чем оплаченных потоков в вашем тарифе, часть запросов начнёт получать ERROR_NO_SLOT_AVAILABLE — держите max_workers в пределах тарифа.
Асинхронная очередь на asyncio
Когда выбрать этот вариант
Для проектов на aiohttp или FastAPI логичнее не заводить системные потоки, а ограничить параллелизм семафором. asyncio.Semaphore пропускает не больше max_concurrent задач одновременно, а asyncio.gather с return_exceptions=True не прерывает всю партию из-за одной ошибки.
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())
Такая очередь особенно уместна, если у вас уже есть асинхронный краулер: CAPTCHA решается в фоне, пока остальной код продолжает обрабатывать другие страницы, а не простаивает в ожидании ответа.
Модель производитель-потребитель
Когда применять этот паттерн
Когда страницы с CAPTCHA обнаруживаются на лету — например, парсер сам находит новые ссылки в процессе обхода — фиксированный список задач не подходит. Паттерн «производитель — потребитель» разделяет источник задач и обработчиков: producer кладёт новые задачи в asyncio.Queue по мере появления, а фиксированное число consumer-ов решает их независимо от темпа парсинга.
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())
Сигнал None в конце очереди — простой способ остановить consumer-ов, когда источник задач исчерпан. При действительно непрерывном парсинге очередь можно вообще не закрывать и держать consumer-ов постоянно активными, добавляя задачи producer-ом по мере обнаружения новых страниц.
Приоритетная очередь
Когда нужна приоритизация
Не все CAPTCHA одинаково важны. Токен для страницы оформления заказа блокирует реальную покупку прямо сейчас, а CAPTCHA на второстепенной информационной странице может подождать. asyncio.PriorityQueue решает задачи в порядке приоритета — меньшее число означает более высокий приоритет.
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())
В примере checkout получает приоритет 1 и решается первым, даже если он попал в очередь позже карточек товара с приоритетом 5. Информационные страницы с приоритетом 10 дожидаются своей очереди последними — очередь не гарантирует порядок отправки, только порядок обработки.
Мониторинг и отчетность
Что отслеживать
Без метрик сложно понять, стоит ли увеличивать max_workers или очередь уже упирается в лимит тарифа. QueueMetrics считает долю успешных решений, среднее время решения и пропускную способность в задачах в минуту — этого достаточно для алертов и простого дашборда.
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"
)
Если очередь логирует URL страниц и токены вместе с пользовательскими данными, храните только то, что вы вправе обрабатывать: для аудитории в РФ ориентируйтесь на 152-ФЗ «О персональных данных», для трансграничных проектов — на подход в духе GDPR. Это вопрос дисциплины логирования, а не возможностей самой очереди.
Типичные проблемы и их решение
Большинство сбоев очереди в продакшене сводятся к пяти причинам ниже.
| Симптом | Причина | Решение |
|---|---|---|
| Очередь растёт, но задачи не решаются | Слишком много воркеров перегружают API | Уменьшите max_workers / max_concurrent |
ERROR_NO_SLOT_AVAILABLE |
Превышен лимит параллельных потоков по тарифу | Добавьте задержку между отправками или уменьшите число воркеров |
| Задачи зависают в очереди | Поток-воркер упал с необработанным исключением | Оберните тело воркера в try/except |
| Память растёт со временем | Результаты не забираются из result_queue |
Периодически вызывайте get_results() |
| Асинхронная очередь блокируется | Пропущен await у одного из вызовов |
Проверьте, что все асинхронные вызовы ожидаются |
Часто задаваемые вопросы
Сколько воркеров ставить, чтобы не словить ERROR_NO_SLOT_AVAILABLE?
Ориентируйтесь на число потоков в вашем тарифе CaptchaAI, а не на случайное число: например, при ADVANCE ($90/мес, 50 потоков) значение max_workers/max_concurrent в районе 40–50 обычно не упирается в лимит. Если ERROR_NO_SLOT_AVAILABLE всё же появляется, добавьте небольшую задержку между отправками или уменьшите параллелизм.
threading или asyncio: что выбрать для очереди CAPTCHA?
Если парсер уже синхронный и переписывать его целиком нет времени — берите очередь на threading, она подключается поверх существующего кода без переделки архитектуры. Для нового проекта или уже асинхронного краулера на aiohttp/FastAPI asyncio-очередь эффективнее: она не создаёт лишних системных потоков и лучше масштабируется на большое число одновременных задач.
Можно ли решать в одной очереди reCAPTCHA v2, Cloudflare Turnstile и GeeTest v3?
Да — очередь агностична к типу CAPTCHA. method и параметры (googlekey/sitekey/pageurl) передаются в submit() отдельно для каждой задачи, а _solve() работает с любым поддерживаемым типом через одни и те же in.php/res.php. Главное — не смешивать параметры разных типов в одном вызове.
Как логировать результаты очереди с учётом персональных данных?
Логируйте URL, ID задачи и статус решения, но не сохраняйте персональные данные пользователей вместе с токеном дольше, чем нужно для отладки. Ориентируйтесь на 152-ФЗ для аудитории в РФ и на подход в духе GDPR для трансграничных проектов: собирайте только то, что вы вправе обрабатывать.
Как масштабировать очередь при росте объёма — с сотен до тысяч CAPTCHA в час?
Сначала поднимите тариф: ADVANCE (50 потоков) или PREMIUM ($170/мес, 100 потоков) обычно хватает на сотни задач в час, CORPORATE ($240/мес, 150 потоков) или ENTERPRISE ($300/мес, 200 потоков) — для более плотного потока. Затем масштабируйте max_workers/max_concurrent вслед за оплаченными потоками и следите за success_rate в QueueMetrics, чтобы вовремя заметить деградацию.
Итоги
Очередь решения CAPTCHA отделяет отправку задач от опроса результатов, позволяя решать десятки и сотни CAPTCHA параллельно через CaptchaAI. Берите многопоточность для синхронного кода, asyncio — для нового или уже асинхронного проекта, производитель-потребитель — для непрерывного парсинга, а приоритетную очередь — там, где порядок решения важен не меньше самого решения.
