Python ThreadPoolExecutor: CAPTCHAs parallel lösen

Mehrere CAPTCHAs lassen sich in Python parallel lösen, ohne die gesamte Codebasis auf asyncio umzustellen – ThreadPoolExecutor aus concurrent.futures genügt. Das Lösen eines CAPTCHAs ist fast reine Wartezeit: Ihr Code hängt an der HTTP-Antwort von CaptchaAI, nicht an der CPU. Genau diese Wartezeit füllen Threads sinnvoll aus. Während ein Thread auf ein Ergebnis wartet, schicken die anderen längst neue Anfragen los.

Wer 20, 200 oder 2.000 Aufgaben nacheinander abarbeitet, summiert diese Wartezeiten zu Minuten oder Stunden. Ein Thread-Pool bringt denselben Stapel in einem Bruchteil der Zeit durch – und das mit synchronem Code, den Sie ohne Umbau in ein bestehendes Selenium- oder Scraping-Projekt einsetzen können.

Warum ThreadPoolExecutor beim CAPTCHA-Lösen effizient ist

Das Lösen von CAPTCHAs ist I/O-lastig: Der Großteil der Zeit vergeht mit dem Warten auf HTTP-Antworten. Python gibt die GIL während dieser I/O-Phasen frei, weshalb Threads hier echte Nebenläufigkeit liefern – anders als bei CPU-lastiger Arbeit. Die folgende Übersicht ordnet die gängigen Ansätze ein:

Für reine I/O-Arbeit ist ThreadPoolExecutor der beste Kompromiss aus Aufwand und Wirkung: kaum Umbau, deutlich mehr Durchsatz.

Grundgerüst: einen CAPTCHA-Stapel parallel abarbeiten

Das Muster ist immer gleich – eine synchrone Solve-Funktion übermittelt eine Aufgabe und fragt anschließend das Ergebnis ab (Polling), der Pool verteilt diese Funktion über mehrere Worker. Das folgende Beispiel löst 20 reCAPTCHA-v2-Aufgaben mit 10 Workern parallel:

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

as_completed liefert die Futures in der Reihenfolge zurück, in der sie fertig werden – nicht in der Reihenfolge der Übermittlung. So verarbeiten Sie jedes Ergebnis, sobald es vorliegt, statt auf den langsamsten Aufruf zu warten.

Verbindungen wiederverwenden: thread-lokale Sessions

Eine neue TCP-Verbindung pro Anfrage kostet unnötig Zeit. Mit einer requests.Session pro Thread bleibt die Verbindung offen und wird über einen HTTPAdapter gepoolt. Thread-lokaler Speicher stellt sicher, dass sich die Worker keine Session teilen:

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

Bei großen Stapeln spart das spürbar Latenz, weil der TLS-Handshake pro Thread nur einmal statt bei jeder Anfrage anfällt.

map() für einfache Stapelverarbeitung

Wenn Sie keine Fehlerbehandlung pro Aufgabe benötigen, ist executor.map() die kompakteste Variante. Ein Wrapper fängt Ausnahmen ab und liefert für jede Aufgabe ein Ergebnis-Dictionary zurück:

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

map() behält die Reihenfolge der Eingabeliste bei – praktisch, wenn Ergebnisse und Aufgaben eindeutig zusammenpassen müssen.

Timeouts absichern

Ein hängender Thread darf nicht den ganzen Pool blockieren. Setzen Sie deshalb zwei Grenzen: ein globales Timeout über as_completed(..., timeout=600) und ein Timeout pro Aufgabe über future.result(timeout=120).

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

Fortschritt in Echtzeit verfolgen

Bei großen Stapeln hilft eine Fortschrittsanzeige. Ein threading.Lock schützt den gemeinsamen Zähler, damit sich parallele Worker beim Hochzählen nicht in die Quere kommen:

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

Wie viele Worker? max_workers an den Tarif koppeln

max_workers bestimmt, wie viele CAPTCHAs gleichzeitig in Bearbeitung sind. Der entscheidende Punkt für die Praxis: CaptchaAI rechnet pro gleichzeitigem Thread ab – nicht pro Lösung. Ihr max_workers-Wert sollte deshalb die Zahl der gebuchten Threads nicht überschreiten, sonst laufen zusätzliche Anfragen nur in eine Warteschlange.

Konkret heißt das: Mit dem Tarif BASIC (15 $/Monat, 5 Threads) setzen Sie max_workers=5, mit STANDARD (30 $/Monat, 15 Threads) bis zu 15 und mit ADVANCE (90 $/Monat, 50 Threads) bis zu 50. Beginnen Sie mit 10 Workern und steigern Sie schrittweise, während Sie die Fehlerraten beobachten.

Der tatsächliche Durchsatz hängt vom CAPTCHA-Typ ab: Ein Bild-CAPTCHA ist in unter 0,5 Sekunden gelöst, reCAPTCHA v2 kann bis zu 60 Sekunden dauern. Derselbe Pool bringt bei einfachen Typen also ein Vielfaches an Lösungen pro Stunde. Auf einem kleinen Hetzner- oder netcup-VPS ist dabei nicht die CPU der Engpass, sondern Ihre Thread-Zuteilung bei CaptchaAI – die Worker warten die meiste Zeit ohnehin nur auf Antworten.

ThreadPoolExecutor oder asyncio?

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

Nehmen Sie ThreadPoolExecutor, wenn:

• Ihre bestehende Codebasis synchron ist
• Sie Bibliotheken einsetzen, die kein Async unterstützen (Selenium, manche ORMs)
• Sie schnelle Parallelität ohne Umbau wollen

Nehmen Sie asyncio, wenn:

• Sie von Grund auf neu bauen
• maximale Effizienz zählt (weniger Betriebssystem-Threads)
• Sie bereits in einem Async-Framework arbeiten (FastAPI, aiohttp)

Häufige Probleme beim Threading

Häufige Fragen

Wie viele Worker passen zu meinem CaptchaAI-Tarif?

So viele, wie Ihr Tarif Threads umfasst. CaptchaAI rechnet pro gleichzeitigem Thread ab, jeder Thread löst im Monat unbegrenzt viele CAPTCHAs. BASIC (15 $/Monat) bietet 5 Threads, STANDARD (30 $/Monat) 15, ADVANCE (90 $/Monat) 50. Setzen Sie max_workers höchstens auf diese Zahl.

Muss ich für Selenium-Projekte auf asyncio umsteigen?

Nein. Selenium ist synchron und lässt sich nicht sauber mit asyncio kombinieren. Genau dafür ist ThreadPoolExecutor gedacht: Sie behalten Ihren bestehenden synchronen Code und gewinnen trotzdem Parallelität.

Was tun bei gehäuften ConnectionError-Meldungen?

Reduzieren Sie max_workers und aktivieren Sie Connection-Pooling über eine thread-lokale requests.Session. Meist entstehen die Fehler, weil mehr Verbindungen geöffnet werden, als der Pool oder das Netzwerk verkraften.

Blockiert die GIL die parallele Ausführung?

Nicht bei I/O-lastiger Arbeit. Während HTTP-Anfragen und time.sleep gibt Python die GIL frei, sodass die Threads echt nebenläufig laufen. Nur bei CPU-lastiger Arbeit begrenzt die GIL die Parallelität – beim CAPTCHA-Lösen ist das nicht der Fall.

Verwandte Leitfäden

• Parallele vs. sequenzielle CAPTCHA-Lösung
• Batch-CAPTCHAs mit Promise.allSettled in Node.js
• Mehrere CAPTCHA-Aufgaben im Stapel lösen