Architekturmuster für CAPTCHA-Solving bei hohem Volumen

Welches Architekturmuster Sie für das CAPTCHA-Solving brauchen, entscheidet vor allem eine Kennzahl: die Abfragen pro Stunde. Bei einigen hundert Lösungen pro Stunde genügt ein schlanker Worker-Pool. Ab mehreren Tausend brauchen Sie eine Warteschlange mit Backpressure, ab mehreren Zehntausend verteilte Warteschlangen über mehrere Maschinen.

Dieser Leitfaden ordnet fünf Muster diesen Volumenstufen zu – vom einfachen Thread-Pool bis zum Multi-Provider-Failover, jeweils mit lauffähigem Python-Code gegen die CaptchaAI-API.

Ein Hinweis vorab: Mehr Architektur ist nicht automatisch besser. Solange die Warteschlangen kurz und die Lösungszeit stabil bleiben, kostet zusätzliche Komplexität nur Wartungsaufwand – ihr Nutzen entsteht erst, wenn ein einzelner Engpass den gesamten Solve-Workflow ausbremst.

Welches Muster passt zu Ihrem Durchsatz?

Muster 1: Worker-Pool für den Einstieg

Ein ThreadPoolExecutor mit 5–20 Worker deckt diesen Bereich ab, ganz ohne eigene Infrastruktur. Jeder Worker übermittelt eine Abfrage an in.php, wartet kurz und fragt das Ergebnis über res.php ab.

• Geeignet für: 100–1.000 Lösungen pro Stunde
• Warum es trägt: Das Lösen ist I/O-lastig – die meiste Zeit vergeht mit Warten –, deshalb skaliert das Muster weit, bevor die Threads selbst zum Engpass werden.

┌──────────┐     ┌──────────────┐     ┌────────────┐
│  Scraper  │────▶│  Thread Pool │────▶│ CaptchaAI  │
│  Tasks    │     │  (5-20       │     │   API      │
│           │◀────│   workers)   │◀────│            │
└──────────┘     └──────────────┘     └────────────┘

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


class SimpleWorkerPool:
    def __init__(self, api_key, max_workers=10):
        self.api_key = api_key
        self.max_workers = max_workers
        self.base = "https://ocr.captchaai.com"

    def _solve_one(self, params):
        params["key"] = self.api_key
        params["json"] = 1

        resp = requests.post(f"{self.base}/in.php", data=params).json()
        if resp["status"] != 1:
            return {"error": resp["request"]}

        task_id = resp["request"]
        time.sleep(10)

        for _ in range(60):
            result = requests.get(
                f"{self.base}/res.php",
                params={"key": self.api_key, "action": "get", "id": task_id, "json": 1},
            ).json()

            if result["request"] == "CAPCHA_NOT_READY":
                time.sleep(5)
                continue
            if result["status"] == 1:
                return {"token": result["request"]}
            return {"error": result["request"]}

        return {"error": "timeout"}

    def solve_batch(self, tasks):
        """tasks: list of (identifier, params) tuples."""
        results = {}
        with ThreadPoolExecutor(max_workers=self.max_workers) as pool:
            futures = {
                pool.submit(self._solve_one, params): ident
                for ident, params in tasks
            }
            for future in as_completed(futures):
                ident = futures[future]
                try:
                    results[ident] = future.result()
                except Exception as e:
                    results[ident] = {"error": str(e)}
        return results

Muster 2: Queue-Pipeline mit Backpressure

Mit steigendem Volumen wird die Entkopplung von Übermittlung und Polling wichtig. Eine beschränkte submit_queue (maxsize=100) erzeugt Backpressure: Sobald die Pipeline gesättigt ist, blockieren die Produzenten, statt die API mit immer neuen Anfragen zu überfluten.

• Geeignet für: 1.000–10.000 Lösungen pro Stunde
• Kernidee: getrennte Submit- und Poll-Worker, damit langsame Ergebnisse das Einreichen neuer Abfragen nicht ausbremsen.

┌────────┐     ┌───────────┐     ┌──────────┐     ┌───────────┐     ┌────────┐
│Producer│────▶│ Submit    │────▶│ Pending  │────▶│  Poll     │────▶│Results │
│        │     │ Queue     │     │ Queue    │     │  Workers  │     │ Queue  │
└────────┘     └───────────┘     └──────────┘     └───────────┘     └────────┘

import queue
import threading
import time
import requests


class QueuePipeline:
    def __init__(self, api_key, submit_workers=5, poll_workers=10):
        self.api_key = api_key
        self.base = "https://ocr.captchaai.com"
        self.submit_queue = queue.Queue(maxsize=100)
        self.pending_queue = queue.Queue()
        self.results = {}
        self.results_lock = threading.Lock()
        self._running = False
        self.submit_workers = submit_workers
        self.poll_workers = poll_workers

    def start(self):
        self._running = True
        for _ in range(self.submit_workers):
            threading.Thread(target=self._submit_worker, daemon=True).start()
        for _ in range(self.poll_workers):
            threading.Thread(target=self._poll_worker, daemon=True).start()

    def stop(self):
        self._running = False

    def add(self, ident, params):
        self.submit_queue.put((ident, params))

    def get_result(self, ident, timeout=300):
        deadline = time.time() + timeout
        while time.time() < deadline:
            with self.results_lock:
                if ident in self.results:
                    return self.results.pop(ident)
            time.sleep(1)
        return {"error": "timeout"}

    def _submit_worker(self):
        while self._running:
            try:
                ident, params = self.submit_queue.get(timeout=1)
            except queue.Empty:
                continue
            params["key"] = self.api_key
            params["json"] = 1
            try:
                resp = requests.post(f"{self.base}/in.php", data=params).json()
                if resp["status"] == 1:
                    self.pending_queue.put((ident, resp["request"], time.time()))
                else:
                    with self.results_lock:
                        self.results[ident] = {"error": resp["request"]}
            except Exception as e:
                with self.results_lock:
                    self.results[ident] = {"error": str(e)}

    def _poll_worker(self):
        while self._running:
            try:
                ident, task_id, submitted_at = self.pending_queue.get(timeout=1)
            except queue.Empty:
                continue

            # Wait at least 10s from submission
            wait = 10 - (time.time() - submitted_at)
            if wait > 0:
                time.sleep(wait)

            try:
                resp = requests.get(
                    f"{self.base}/res.php",
                    params={"key": self.api_key, "action": "get", "id": task_id, "json": 1},
                ).json()

                if resp["request"] == "CAPCHA_NOT_READY":
                    self.pending_queue.put((ident, task_id, submitted_at))
                    time.sleep(3)
                elif resp["status"] == 1:
                    with self.results_lock:
                        self.results[ident] = {"token": resp["request"]}
                else:
                    with self.results_lock:
                        self.results[ident] = {"error": resp["request"]}
            except Exception:
                self.pending_queue.put((ident, task_id, submitted_at))
                time.sleep(5)

Verwendung:

pipeline = QueuePipeline("YOUR_API_KEY")
pipeline.start()

# Add CAPTCHAs
pipeline.add("page_1", {"method": "turnstile", "sitekey": "KEY", "pageurl": "URL"})
pipeline.add("page_2", {"method": "userrecaptcha", "googlekey": "KEY", "pageurl": "URL"})

# Get results
result1 = pipeline.get_result("page_1")
result2 = pipeline.get_result("page_2")

pipeline.stop()

Muster 3: Circuit Breaker gegen kaskadierende Fehler

Fällt die Lösungs-API kurzzeitig aus oder antwortet sie langsam, verschlimmern blinde Wiederholungen die Lage nur. Ein Circuit Breaker weist Anfragen nach mehreren Fehlern in Folge sofort ab, statt sie in Timeouts laufen zu lassen.

• Geeignet für: instabile oder überlastete API-Phasen
• Ablauf: Nach einer Abkühlzeit (reset_timeout) prüft der halboffene Zustand mit einer einzelnen Anfrage, ob sich die API erholt hat – erst dann schließt der Kreis wieder.

import time
import threading


class CircuitBreaker:
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing — reject requests
    HALF_OPEN = "half_open"  # Testing recovery

    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.state = self.CLOSED
        self.failure_count = 0
        self.last_failure_time = 0
        self.lock = threading.Lock()

    def can_proceed(self):
        with self.lock:
            if self.state == self.CLOSED:
                return True
            if self.state == self.OPEN:
                if time.time() - self.last_failure_time > self.reset_timeout:
                    self.state = self.HALF_OPEN
                    return True
                return False
            # HALF_OPEN: allow one request
            return True

    def record_success(self):
        with self.lock:
            self.failure_count = 0
            self.state = self.CLOSED

    def record_failure(self):
        with self.lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = self.OPEN


class ResilientSolver:
    def __init__(self, api_key):
        self.api_key = api_key
        self.breaker = CircuitBreaker(failure_threshold=5, reset_timeout=60)

    def solve(self, params):
        if not self.breaker.can_proceed():
            raise Exception("Circuit open — API degraded, try later")

        try:
            result = self._do_solve(params)
            self.breaker.record_success()
            return result
        except Exception as e:
            self.breaker.record_failure()
            raise

    def _do_solve(self, params):
        # Standard solve logic
        pass

Muster 4: Token-Puffer für latenzkritische Pfade

Auf latenzkritischen Pfaden – etwa wenn ein Nutzer im Frontend auf die Antwort wartet – stört selbst eine Lösungszeit von 10–15 Sekunden. Ein kleiner Puffer frisch gelöster Token, kontinuierlich nachgefüllt, stellt sofort ein Token bereit.

• Geeignet für: Frontend-Pfade, auf denen jemand aktiv auf die Antwort wartet
• Wichtig: Token laufen schnell ab (oft binnen rund 120 Sekunden). Der Puffer arbeitet deshalb mit kurzer TTL und verwirft abgelaufene Einträge automatisch – es geht um das Glätten von Latenzspitzen, nicht um das Anhäufen von Token.

import queue
import threading
import time


class TokenBuffer:
    def __init__(self, solver, params, buffer_size=5, ttl_seconds=90):
        self.solver = solver
        self.params = params
        self.buffer = queue.Queue(maxsize=buffer_size)
        self.ttl = ttl_seconds
        self.buffer_size = buffer_size
        self._running = False

    def start(self):
        self._running = True
        threading.Thread(target=self._fill_loop, daemon=True).start()

    def stop(self):
        self._running = False

    def get_token(self, timeout=30):
        """Get a pre-solved token. Returns None if buffer empty."""
        try:
            token, created_at = self.buffer.get(timeout=timeout)
            if time.time() - created_at > self.ttl:
                # Token expired, try next
                return self.get_token(timeout=timeout)
            return token
        except queue.Empty:
            return None

    def _fill_loop(self):
        while self._running:
            if self.buffer.qsize() < self.buffer_size:
                try:
                    token = self.solver.solve(self.params)
                    self.buffer.put((token, time.time()))
                except Exception:
                    time.sleep(5)
            else:
                time.sleep(2)

Muster 5: Multi-Provider-Failover

Ab sehr hohem Volumen darf kein einzelner Anbieter zum Single Point of Failure werden. Der MultiProviderSolver sortiert die Anbieter nach Priorität und leitet bei einem Ausfall auf den nächsten weiter.

• Geeignet für: verteilte Aufbauten mit mehreren Anbietern
• Kernidee: Jeder Provider erhält einen eigenen Circuit Breaker, damit der Durchsatz stabil bleibt, wenn einer vorübergehend gedrosselt wird oder ausfällt.

class MultiProviderSolver:
    def __init__(self, providers):
        """providers: list of (name, solver_instance, priority) tuples."""
        self.providers = sorted(providers, key=lambda x: x[2])
        self.breakers = {name: CircuitBreaker() for name, _, _ in providers}

    def solve(self, params):
        errors = []
        for name, solver, _ in self.providers:
            if not self.breakers[name].can_proceed():
                continue
            try:
                result = solver.solve(params)
                self.breakers[name].record_success()
                return result
            except Exception as e:
                self.breakers[name].record_failure()
                errors.append(f"{name}: {e}")
        raise Exception(f"All providers failed: {'; '.join(errors)}")

Parallelität und Thread-Kontingent

Jeder gleichzeitig laufende Solve belegt bei CaptchaAI genau einen Thread. Die Parallelität Ihrer Architektur muss also zum gebuchten Thread-Kontingent passen: CaptchaAI rechnet Thread-basiert ab – pro gleichzeitigem Thread, nicht pro Lösung – mit unbegrenzten Lösungen je Thread im Abrechnungsmonat.

Ein Worker-Pool mit 50 gleichzeitigen Anfragen braucht entsprechend einen Plan mit mindestens 50 Threads, etwa ADVANCE (90 $/Monat, 50 Threads). Für eine Queue-Pipeline mit rund 100 parallelen Solves passt PREMIUM (170 $/Monat, 100 Threads). Verteilte Aufbauten im Bereich mehrerer Tausend paralleler Abfragen liegen bei VIP-1 (1.500 $/Monat, 1.000 Threads) und darüber. Kleinere Projekte starten mit BASIC (15 $/Monat, 5 Threads) oder STANDARD (30 $/Monat, 15 Threads). Die Preise sind in US-Dollar.

Skalierungsrichtlinien: Volumen, Architektur, Threads

Betrieb in der DACH-Praxis

In der Praxis laufen solche Pipelines häufig auf günstigen VPS- oder Cloud-Instanzen. Anbieter wie Hetzner, netcup oder IONOS eignen sich gut, um Worker über mehrere Maschinen zu verteilen; die Warteschlange liegt dann zentral in Redis oder RabbitMQ. Deployments automatisieren Sie über GitLab CI oder GitHub Actions, wie sie in vielen DACH-Teams ohnehin im Einsatz sind.

Bedient Ihre Pipeline Web-Scraping, sollten Sie den datenschutzrechtlichen Rahmen mitdenken: IP-Adressen und weitere personenbezogene Daten fallen unter die DSGVO. Prüfen Sie Rechtsgrundlage und Datenflüsse Ihrer eigenen Verarbeitung – unabhängig vom Lösungsdienst.

Monitoring und Erfolgsquote im Betrieb

Ab mehreren Tausend Lösungen pro Stunde führt kein Weg an Monitoring vorbei. Erfassen Sie pro CAPTCHA-Typ Versuche, Erfolge und Lösungszeit. Daraus ergeben sich die entscheidenden Kennzahlen: Erfolgsquote je Typ und die Perzentile der Lösungszeit. Fällt die Quote oder steigen die Zeiten, ist das ein Frühwarnsignal, bevor der Circuit Breaker greift.

import logging
from collections import defaultdict

logger = logging.getLogger("captcha_scale")


class ScaleMetrics:
    def __init__(self):
        self.counts = defaultdict(int)
        self.times = defaultdict(list)

    def record(self, captcha_type, success, elapsed):
        key = f"{captcha_type}_{'ok' if success else 'fail'}"
        self.counts[key] += 1
        self.times[captcha_type].append(elapsed)

    def report(self):
        for captcha_type in set(k.rsplit("_", 1)[0] for k in self.counts):
            ok = self.counts.get(f"{captcha_type}_ok", 0)
            fail = self.counts.get(f"{captcha_type}_fail", 0)
            total = ok + fail
            rate = (ok / total * 100) if total else 0
            times = self.times.get(captcha_type, [])
            avg_time = sum(times) / len(times) if times else 0
            logger.info(
                f"{captcha_type}: {total} solves, {rate:.1f}% success, {avg_time:.1f}s avg"
            )

Häufige Fragen

Welchen CaptchaAI-Plan brauche ich für 10.000 CAPTCHAs pro Stunde?

Das hängt von der nötigen Parallelität ab, nicht von der Gesamtzahl. Bei einer Lösungszeit von rund 10–15 Sekunden je Abfrage genügen für dieses Volumen oft 50–100 gleichzeitige Threads – also ADVANCE (90 $/Monat, 50 Threads) bis PREMIUM (170 $/Monat, 100 Threads). Da jeder Thread unbegrenzt viele Lösungen pro Monat erlaubt, begrenzt allein die Parallelität den Durchsatz.

Ab welchem Durchsatz lohnt sich eine Queue-basierte Pipeline?

Etwa ab 1.000 Lösungen pro Stunde. Darunter reicht ein Worker-Pool. Sobald Sie Übermittlung und Polling entkoppeln und Backpressure brauchen, um die API nicht zu überlasten, spielt Muster 2 seine Stärken aus.

Was tun bei ERROR_NO_SLOT_AVAILABLE?

Dieser Fehler zeigt an, dass alle Threads Ihres Plans belegt sind. Reduzieren Sie die Parallelität über eine beschränkte Warteschlange, bremsen Sie mit Rate-Limiting und einem Circuit Breaker – oder buchen Sie ein Plan-Tier mit mehr Threads. Beobachten Sie die Fehlerquote dabei laufend im Monitoring.

Redis oder RabbitMQ für verteilte Warteschlangen?

Beides funktioniert. Redis ist schnell aufgesetzt und reicht für viele Pipelines; RabbitMQ bietet ausgereiftere Zustellgarantien und flexibleres Routing für komplexe Topologien. Ab mehreren Maschinen (10.000+ pro Stunde) ist eine zentrale Queue in jedem Fall sinnvoller als prozesslokale Threads.

Wie überwache ich Erfolgsquote und Lösungszeit im Betrieb?

Erfassen Sie pro CAPTCHA-Typ Anzahl, Erfolge und Lösungszeit – so wie in der ScaleMetrics-Klasse oben. Aussagekräftig sind die Erfolgsquote je Typ und die Perzentile der Lösungszeit (Median, P90, P99). Sinkt die Quote oder steigen die Zeiten, greift der Circuit Breaker, bevor Fehler kaskadieren.

Verwandte Leitfäden

• Von den Grundlagen bis in die Produktion
• API-Kurzreferenz

Skalieren Sie Ihre Solve-Pipeline auf jedes Volumen – jetzt mit CaptchaAI starten.