ThreadPoolExecutor в Python для параллельного решения CAPTCHA

Если код отправляет CAPTCHA в API CaptchaAI по одной, а нужно обрабатывать сразу десятки задач — необязательно переписывать всё на asyncio. Решение CAPTCHA — это I/O-bound операция: почти всё время уходит на ожидание ответа HTTP API, а не на вычисления, и во время такого ожидания Python освобождает GIL. Поэтому concurrent.futures.ThreadPoolExecutor из стандартной библиотеки даёт реальный параллелизм поверх обычного синхронного кода — без переписывания архитектуры и без новых зависимостей. Модуль одинаково хорошо подходит и для парсинга каталогов с защитой Cloudflare, и для QA-тестирования формы регистрации, и для пакетного прогона сотен задач через CaptchaAI ночью, пока команда спит.

Базовая реализация

Ниже — рабочий пример: функция solve_captcha отправляет задачу в in.php, затем опрашивает res.php, пока не получит результат или не истечёт тайм-аут. Дальше она просто передаётся в пул потоков как обычная синхронная функция:

import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

API_KEY = os.environ["CAPTCHAAI_API_KEY"]


def solve_captcha(sitekey, pageurl):
    """Synchronous CAPTCHA solve — submit and poll."""
    # Submit
    resp = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": 1
    })
    data = resp.json()

    if data.get("status") != 1:
        raise RuntimeError(data.get("request", "Submit failed"))

    captcha_id = data["request"]

    # Poll for result
    for _ in range(60):
        time.sleep(5)
        result = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY,
            "action": "get",
            "id": captcha_id,
            "json": 1
        }).json()

        if result.get("status") == 1:
            return result["request"]
        if result.get("request") != "CAPCHA_NOT_READY":
            raise RuntimeError(result.get("request", "Unknown error"))

    raise TimeoutError("Solve timeout after 300s")


# Batch solve with ThreadPoolExecutor
tasks = [
    {"sitekey": "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-", "pageurl": f"https://example.com/page/{i}"}
    for i in range(20)
]

start = time.time()

with ThreadPoolExecutor(max_workers=10) as executor:
    futures = {
        executor.submit(solve_captcha, t["sitekey"], t["pageurl"]): t
        for t in tasks
    }

    solved = 0
    failed = 0

    for future in as_completed(futures):
        task = futures[future]
        try:
            solution = future.result()
            solved += 1
            print(f"[OK] {task['pageurl']}: {solution[:30]}...")
        except Exception as e:
            failed += 1
            print(f"[ERR] {task['pageurl']}: {e}")

elapsed = time.time() - start
print(f"\nDone: {solved} solved, {failed} failed in {elapsed:.1f}s")

Сколько потоков задавать в max_workers

Значение max_workers определяет, сколько задач CaptchaAI обрабатывает у вас одновременно. Слишком маленькое число оставляет пропускную способность на столе; слишком большое — упирается в лимит соединений на стороне вашей сети или в лимит потоков вашего тарифа CaptchaAI. Начинайте с 10 и поднимайте значение постепенно, следя за долей ошибок ConnectionError в логах:

Совет: если тариф CaptchaAI ограничивает число одновременных потоков, не ставьте max_workers выше этого лимита — лишние потоки просто встанут в очередь и не ускорят пакет.

Переиспользуйте соединение через requests.Session

Новое TCP-соединение на каждый запрос — это лишняя задержка, которая особенно заметна при десятках параллельных потоков. Создайте requests.Session один раз на поток через threading.local и переиспользуйте её при каждом обращении к in.php и res.php:

import threading

# Thread-local storage for sessions
thread_local = threading.local()


def get_session():
    """Get or create a thread-local session."""
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
        # Configure connection pooling
        adapter = requests.adapters.HTTPAdapter(
            pool_connections=10,
            pool_maxsize=10,
            max_retries=2
        )
        thread_local.session.mount("https://", adapter)
    return thread_local.session


def solve_captcha_pooled(sitekey, pageurl):
    """Solve using thread-local connection pooling."""
    session = get_session()

    resp = session.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": 1
    })
    data = resp.json()

    if data.get("status") != 1:
        raise RuntimeError(data.get("request"))

    captcha_id = data["request"]

    for _ in range(60):
        time.sleep(5)
        result = session.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY,
            "action": "get",
            "id": captcha_id,
            "json": 1
        }).json()

        if result.get("status") == 1:
            return result["request"]
        if result.get("request") != "CAPCHA_NOT_READY":
            raise RuntimeError(result.get("request"))

    raise TimeoutError("Solve timeout")

Типичные проблемы при пакетном решении

• Потоки как будто зависли. Каждый ждёт time.sleep во время опроса — это ожидаемо, поток отдаёт GIL другим во время sleep.
• Всплеск ConnectionError. Слишком много соединений сразу — уменьшите max_workers и включите пул через Session.
• Результаты приходят не по порядку. as_completed отдаёт футуры по завершении, а не по отправке — для порядка используйте map() или сопоставляйте через словарь.
• Растёт потребление памяти. Крупные объекты результатов копятся в futures — обрабатывайте их сразу в цикле as_completed, не храните все разом.

Пакетная обработка через executor.map()

Когда отдельная обработка ошибок для каждой задачи не нужна, map() даёт тот же результат более коротким кодом.

Python сам разбирает список задач по пулу и возвращает результаты в исходном порядке — писать вручную словарь futures не нужно:

def solve_task(task):
    """Wrapper that returns result dict."""
    try:
        solution = solve_captcha_pooled(task["sitekey"], task["pageurl"])
        return {"url": task["pageurl"], "solution": solution, "error": None}
    except Exception as e:
        return {"url": task["pageurl"], "solution": None, "error": str(e)}


with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(solve_task, tasks))

solved = [r for r in results if r["solution"]]
failed = [r for r in results if r["error"]]
print(f"Solved: {len(solved)}, Failed: {len(failed)}")

Как не дать одному зависшему потоку сорвать весь пул

Одна CAPTCHA, которая не решается и не отдаёт ошибку, не должна держать всю партию. Задайте два уровня тайм-аута — общий для пула и отдельный для каждой задачи:

from concurrent.futures import TimeoutError as FuturesTimeout

with ThreadPoolExecutor(max_workers=10) as executor:
    futures = {
        executor.submit(solve_captcha_pooled, t["sitekey"], t["pageurl"]): t
        for t in tasks
    }

    for future in as_completed(futures, timeout=600):  # 10 min global timeout
        task = futures[future]
        try:
            solution = future.result(timeout=120)  # 2 min per task
            print(f"[OK] {task['pageurl']}")
        except FuturesTimeout:
            print(f"[TIMEOUT] {task['pageurl']}")
        except Exception as e:
            print(f"[ERR] {task['pageurl']}: {e}")

Прогресс-бар для пакетного решения

Для длинных пакетов полезно видеть, сколько задач уже готово, не дожидаясь конца всего прогона. Добавьте счётчик под блокировкой, чтобы потоки не портили друг другу вывод:

import threading

progress_lock = threading.Lock()
progress = {"done": 0, "total": 0}


def solve_with_progress(task):
    result = solve_task(task)
    with progress_lock:
        progress["done"] += 1
        pct = progress["done"] / progress["total"] * 100
        print(f'\r  Progress: {progress["done"]}/{progress["total"]} ({pct:.0f}%)', end="")
    return result


progress["total"] = len(tasks)

with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(solve_with_progress, tasks))

print()  # Newline after progress

ThreadPoolExecutor или asyncio: что выбрать

ThreadPoolExecutor встраивается в существующий синхронный код за пять минут. asyncio требует переписать всю цепочку вызовов как async, зато экономит системные ресурсы за счёт меньшего числа потоков ОС — актуально, если вы и так строите сервис на FastAPI или aiohttp:

# ThreadPoolExecutor — drop into existing sync code
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(solve_task, tasks))

# asyncio — requires async function chain
async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [solve_async(session, t) for t in task_list]
        results = await asyncio.gather(*tasks)

Частые вопросы

Мешает ли GIL реальному параллелизму?

Нет — на I/O-вызовах вроде HTTP-запросов и time.sleep Python освобождает GIL, и потоки реально работают параллельно на сетевых операциях. GIL ограничивает только CPU-bound нагрузку, а решение CAPTCHA к ней не относится.

Как поймать зависшую задачу, если res.php не отвечает?

Задайте timeout у as_completed() для всего пула и future.result(timeout=...) для отдельной задачи — так один зависший запрос не заблокирует остальные (раздел «Как не дать одному зависшему потоку сорвать весь пул» выше).

Стоит ли заменить ThreadPoolExecutor на ProcessPoolExecutor?

Смысла нет. Решение CAPTCHA — I/O-bound задача, а ProcessPoolExecutor добавляет накладные расходы на межпроцессное взаимодействие без выигрыша в скорости. Для этой нагрузки потоки эффективнее процессов.

Почему тарификация по потокам удобна агентствам, которые берут проекты в USD?

CaptchaAI считает потоки, а не отдельные решения: каждый тариф даёт фиксированное число одновременных потоков с неограниченным числом решений в месяц — от BASIC ($15/мес, 5 потоков) до VIP-3 ($7,500/мес, 5000 потоков). Для фрилансеров и агентств, которые выставляют счета клиентам в валюте, отличной от доллара, плоская месячная ставка в USD предсказуемее, чем оплата за каждое отдельное решение.

На что обратить внимание при пакетном сборе данных с нескольких сайтов?

Параллелизм ускоряет сбор, но не снимает с вас ответственность за то, какие данные вы собираете. Если в задачах фигурируют персональные данные, для аудитории в РФ актуально требование 152-ФЗ «О персональных данных», а для трансграничных проектов — стандарты due diligence в духе GDPR: собирайте только то, что вы вправе обрабатывать. Это вопрос дисциплины вашей команды, а не функция CaptchaAI.

Следующие шаги

Подключите ThreadPoolExecutor к своему пайплайну и получите API-ключ CaptchaAI, чтобы решать десятки задач параллельно уже сегодня:

• CaptchaAI Quickstart: ваше первое решение CAPTCHA за 5 минут
• Как решить reCAPTCHA v2 через API: пошаговое руководство
• Как решить Cloudflare Turnstile через API
• Как решить GeeTest v3 с помощью API