Redis-Warteschlange + CaptchaAI: Verteilte CAPTCHA-Verarbeitung

300 Formulare, vor jedem ein reCAPTCHA v2: Arbeitet ein einzelner Prozess diese Liste nacheinander ab und wartet pro Abfrage bis zu 60 Sekunden auf das Token, läuft der Job über Stunden – obwohl die Abfragen längst parallel laufen könnten.

Die Antwort ist eine Redis-Liste zwischen Anwendung und CaptchaAI. Producer legen Aufgaben ab, ein Pool von Workern holt sie blockierend heraus, übermittelt sie an in.php, fragt das Ergebnis über res.php ab und schreibt das Token in einen Redis-Hash; ein Pub/Sub-Kanal meldet fertige Aufgaben zurück.

Dieser Leitfaden baut das komplette Producer-Consumer-System in Python auf – inklusive der Thread-Planung, die im Betrieb über den Durchsatz entscheidet.

Wann sich eine Warteschlange wirklich lohnt

Solange ein einzelner Prozess ausreicht, bleiben direkte API-Aufrufe die einfachere Wahl. Eine Warteschlange lohnt sich, sobald einer dieser Fälle zutrifft:

• Mehrere Quellen erzeugen Aufgaben – Crawler, Test-Suite und Cronjob teilen sich denselben Worker-Pool.
• Producer und Solver laufen getrennt – niemand blockiert, während eine Abfrage 10 bis 60 Sekunden dauert, und Lastspitzen puffert die Warteschlange ab.
• Priorisierung ist nötig – ein Login-CAPTCHA aus dem laufenden Betrieb darf vor der nächtlichen Scraping-Charge landen; im Code unten erledigt das lpush statt rpush.

Architektur in drei Redis-Strukturen

Drei Bausteine genügen: Eine Liste transportiert Aufgaben nach dem FIFO-Prinzip, ein Hash speichert Ergebnisse, ein Pub/Sub-Kanal benachrichtigt Wartende. Kein Worker kennt einen anderen – skalieren heißt hier: einen weiteren Prozess starten.

Producers → Redis List (FIFO queue) → Workers → CaptchaAI API
                                          ↓
                                   Redis Hash (results)
                                          ↓
                                   Redis Pub/Sub (notifications)

Der Warteschlangen-Manager

CaptchaQueue kapselt vier Operationen: einreihen, abholen, Ergebnis ablegen, Ergebnis lesen. blpop blockiert bis zu 30 Sekunden und liefert die älteste Aufgabe – aktives Polling gegen Redis entfällt. Aufgaben mit priority="high" gehen per lpush nach vorn; ein separater setex-Schlüssel markiert den Ablauf, denn der Hash selbst kennt keine Feld-TTL.

import json
import time
import uuid
import redis


class CaptchaQueue:
    """Redis-backed CAPTCHA task queue."""

    def __init__(self, redis_url="redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
        self.queue_key = "captcha:tasks"
        self.results_key = "captcha:results"
        self.notify_channel = "captcha:done"

    def submit(self, method, params, priority="normal"):
        """Submit a CAPTCHA task to the queue."""
        task_id = str(uuid.uuid4())[:8]
        task = {
            "id": task_id,
            "method": method,
            "params": params,
            "submitted_at": time.time(),
            "priority": priority,
        }

        if priority == "high":
            self.redis.lpush(self.queue_key, json.dumps(task))
        else:
            self.redis.rpush(self.queue_key, json.dumps(task))

        return task_id

    def fetch(self, timeout=30):
        """Fetch next task from queue (blocking)."""
        result = self.redis.blpop(self.queue_key, timeout=timeout)
        if result is None:
            return None
        _, raw = result
        return json.loads(raw)

    def store_result(self, task_id, result):
        """Store task result and notify listeners."""
        self.redis.hset(
            self.results_key,
            task_id,
            json.dumps(result),
        )
        # Notify via pub/sub
        self.redis.publish(self.notify_channel, task_id)
        # Set TTL on result (1 hour)
        # Results are in a hash, so we track expiry separately
        self.redis.setex(
            f"captcha:ttl:{task_id}", 3600, "1",
        )

    def get_result(self, task_id):
        """Get result for a task (non-blocking)."""
        raw = self.redis.hget(self.results_key, task_id)
        if raw:
            return json.loads(raw)
        return None

    def wait_result(self, task_id, timeout=120):
        """Wait for a task result via polling."""
        start = time.time()
        while time.time() - start < timeout:
            result = self.get_result(task_id)
            if result:
                return result
            time.sleep(1)
        return None

    def queue_stats(self):
        """Get queue statistics."""
        return {
            "pending": self.redis.llen(self.queue_key),
            "completed": self.redis.hlen(self.results_key),
        }

Der Worker-Prozess

Der Worker bleibt bewusst simpel: Aufgabe holen, lösen, Ergebnis ablegen, weitermachen. Die Integration steckt in _solve() – ein POST an in.php mit key, method und den typspezifischen Parametern, danach ein Polling-Loop gegen res.php im Fünf-Sekunden-Takt, bis die Antwort nicht mehr CAPCHA_NOT_READY lautet. Fehler verschwinden nicht im Log, sondern landen mit status: error in Redis. Typ und Parameter tauschen Sie je nach Abfrage: userrecaptcha für reCAPTCHA v2 und v3, turnstile für Cloudflare Turnstile, geetest für GeeTest v3, post für Bild- und Rasterbild-CAPTCHAs.

import os
import time
import requests


class QueueWorker:
    """Worker that processes CAPTCHA tasks from Redis queue."""

    def __init__(self, api_key, queue):
        self.api_key = api_key
        self.queue = queue
        self.base = "https://ocr.captchaai.com"

    def run(self):
        """Main worker loop."""
        worker_id = os.getpid()
        print(f"Worker {worker_id} started")

        while True:
            task = self.queue.fetch(timeout=30)
            if task is None:
                continue

            task_id = task["id"]
            print(f"[{worker_id}] Processing {task_id}")

            start = time.time()
            try:
                token = self._solve(task["method"], task["params"])
                duration = time.time() - start
                self.queue.store_result(task_id, {
                    "status": "success",
                    "token": token,
                    "duration": f"{duration:.1f}s",
                })
                print(f"[{worker_id}] {task_id} solved in {duration:.1f}s")

            except Exception as e:
                self.queue.store_result(task_id, {
                    "status": "error",
                    "error": str(e),
                })
                print(f"[{worker_id}] {task_id} failed: {e}")

    def _solve(self, method, params, timeout=120):
        resp = requests.post(f"{self.base}/in.php", data={
            "key": self.api_key,
            "method": method,
            "json": 1,
            **params,
        }, timeout=30)
        result = resp.json()

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

        captcha_id = result["request"]
        start = time.time()

        while time.time() - start < timeout:
            time.sleep(5)
            resp = requests.get(f"{self.base}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": captcha_id,
                "json": 1,
            }, timeout=15)
            data = resp.json()
            if data["request"] != "CAPCHA_NOT_READY":
                if data.get("status") == 1:
                    return data["request"]
                raise RuntimeError(data["request"])

        raise TimeoutError("Solve timeout")


# Run worker
if __name__ == "__main__":
    queue = CaptchaQueue()
    worker = QueueWorker(os.environ["CAPTCHAAI_KEY"], queue)
    worker.run()

Mehrere Worker parallel starten

Für den Anfang reicht multiprocessing: Jeder Prozess baut seine eigene Redis-Verbindung auf und konkurriert über blpop um dieselbe Liste – Redis serialisiert die Zugriffe, doppelte Auslieferung entsteht nicht. Vier Prozesse pro Maschine sind ein guter Startwert; im Produktivbetrieb übernehmen systemd, Docker Compose oder Kubernetes den Neustart abgestürzter Worker.

import multiprocessing
import os


def start_workers(num_workers=4):
    """Launch multiple worker processes."""
    queue = CaptchaQueue()
    processes = []

    for i in range(num_workers):
        p = multiprocessing.Process(
            target=run_worker,
            args=(os.environ["CAPTCHAAI_KEY"],),
        )
        p.start()
        processes.append(p)
        print(f"Started worker {i + 1}/{num_workers}")

    return processes


def run_worker(api_key):
    queue = CaptchaQueue()
    worker = QueueWorker(api_key, queue)
    worker.run()


# Launch
processes = start_workers(num_workers=4)

Producer-Seite: Aufgaben einreihen

Der Producer kennt weder Worker noch API-Schlüssel: Er reiht Aufgaben ein, sammelt die Task-IDs und holt die Ergebnisse ab, wenn er sie braucht. wait_result() fragt den Hash im Sekundentakt ab – für Skripte genug, für Dienste mit vielen offenen Aufgaben ist Pub/Sub sparsamer.

queue = CaptchaQueue()

# Submit tasks
urls = [
    "https://site1.com/login",
    "https://site2.com/register",
    "https://site3.com/checkout",
]

task_ids = []
for url in urls:
    tid = queue.submit("userrecaptcha", {
        "googlekey": "SITE_KEY",
        "pageurl": url,
    })
    task_ids.append(tid)
    print(f"Submitted {tid} for {url}")

# Wait for all results
for tid in task_ids:
    result = queue.wait_result(tid, timeout=120)
    status = result["status"] if result else "timeout"
    print(f"{tid}: {status}")

# Check queue stats
print(queue.queue_stats())

Ergebnisse per Pub/Sub einsammeln

Wer nicht pollen will, abonniert den Benachrichtigungskanal. Der Listener läuft in einem Hintergrund-Thread und liest bei jeder gemeldeten Task-ID das Ergebnis direkt aus dem Hash:

import threading


def listen_results(queue):
    """Listen for completed task notifications."""
    pubsub = queue.redis.pubsub()
    pubsub.subscribe(queue.notify_channel)

    for message in pubsub.listen():
        if message["type"] == "message":
            task_id = message["data"].decode()
            result = queue.get_result(task_id)
            print(f"Task {task_id} completed: {result['status']}")


# Run listener in background
listener = threading.Thread(
    target=listen_results,
    args=(CaptchaQueue(),),
    daemon=True,
)
listener.start()

Pub/Sub stellt nur an verbundene Abonnenten zu; maßgeblich bleibt der Hash, die Benachrichtigung ist die Abkürzung.

Dimensionierung: Threads zählen, nicht Worker

Die häufigste Fehlplanung: Worker hochzählen und sich wundern, dass der Durchsatz nicht mitwächst. CaptchaAI rechnet Thread-basiert ab – nicht pro Lösung. Ein Thread ist eine Abfrage in Bearbeitung; die Lösungen pro Thread sind im Abrechnungsmonat unbegrenzt. Mehr gleichzeitig lösende Worker als Threads bringen nichts außer wartenden Prozessen.

Rechnen Sie von der Lösungszeit rückwärts: Cloudflare Turnstile wird in unter 10 Sekunden gelöst, reCAPTCHA v2 in unter 60 Sekunden. Ein Thread schafft damit rund sechs Turnstile-Abfragen pro Minute, aber höchstens etwa eine reCAPTCHA-v2-Abfrage – für 600 solcher Aufgaben in zehn Minuten brauchen Sie also rund 60 Threads. Preise in US-Dollar.

Betrieb im DACH-Umfeld: Hetzner, GitLab CI und DSGVO

Ein typisches Setup: Ein Team in Köln lässt die nächtliche QA-Suite für den eigenen Shopware-Shop auf einer Hetzner-Cloud-Instanz in Falkenstein laufen – Redis als Container, vier Worker daneben, ausgelöst aus einer GitLab-CI-Pipeline. Der Producer ist ein CI-Job über die eigene Staging-Umgebung; die Worker liefern die Tokens, bevor der Selenium-Teil weiterläuft.

Zwei Punkte gehören hier fest in die Planung: IP-Adressen und Formulardaten gelten nach DSGVO als personenbezogene Daten – prüfen Sie, welche pageurl-Werte und Logzeilen in Redis landen, und setzen Sie kurze TTLs. Und Redis gehört nie ungeschützt ins öffentliche Netz: Binden Sie die Instanz an das interne Netz Ihres Anbieters – Hetzner, IONOS und netcup bieten das an – und vergeben Sie ein Passwort; ein offener Port 6379 kostet im Zweifel den Datenbestand.

Zuverlässigkeit: Abstürze, TTL und erneute Versuche

blpop entfernt die Aufgabe aus der Liste, bevor der Worker sie bearbeitet. Stürzt der Prozess danach ab, ist sie verloren – beim nächtlichen Scraping-Lauf verschmerzbar, bei kundennahen Abläufen nicht. Drei Stellschrauben:

• Zuverlässige Entnahme: brpoplpush verschiebt die Aufgabe in eine Bearbeitungsliste; ein Aufräum-Job holt Einträge zurück, die dort länger als das Timeout liegen.
• Erneute Versuche mit Backoff: einen attempts-Zähler mitführen und Fehlschläge höchstens zwei- bis dreimal erneut einreihen.
• Token-Lebensdauer: Ein gelöstes Token gilt in der Regel rund 120 Sekunden. Lösen Sie die Abfrage deshalb kurz vor der Übermittlung des Formulars.

Fehlerbehebung

Häufige Fragen

Wie viele Worker passen zu meinem CaptchaAI-Plan?

Höchstens so viele, wie Ihr Plan Threads enthält. ADVANCE (90 $/Monat) bringt 50 Threads mit, also lohnen sich maximal 50 gleichzeitig lösende Worker. Die Lösungen pro Thread sind unbegrenzt: Sie bezahlen Parallelität, nicht Menge.

Was passiert, wenn ein Worker mitten in der Bearbeitung abstürzt?

Die Aufgabe ist verloren, weil blpop sie bereits aus der Liste genommen hat. Mit brpoplpush landet sie in einer Bearbeitungsliste, aus der ein Aufräum-Job sie zurückholt.

Wie lange sollten gelöste Tokens in Redis liegen bleiben?

So kurz wie möglich. Tokens laufen typischerweise nach rund 120 Sekunden ab; ein Hash mit einer Stunde TTL dient danach nur noch der Nachvollziehbarkeit. Status und Dauer dürfen bleiben, das Token nicht.

Sind Redis Streams die bessere Wahl als Listen?

Für den Dauerbetrieb ja: Streams bringen Consumer-Gruppen und Bestätigungen mit, ein abgestürzter Consumer verliert damit keine Aufgabe. Für kleine Setups bleiben Listen pragmatisch – umstellen lässt sich später.

Können verschiedene CAPTCHA-Typen in einer Warteschlange laufen?

Ja. Jede Aufgabe trägt ihr eigenes method-Feld, das der Worker unverändert an in.php weiterreicht – Bild-CAPTCHAs, reCAPTCHA v2 und v3, Turnstile und GeeTest v3 laufen gemischt durch denselben Pool. Getrennte Listen lohnen erst, wenn schnelle Typen hinter langsamen warten.

Verwandte Leitfäden

• Message-Queue-Integration mit RabbitMQ
• Mehrere CAPTCHAs in einer Charge lösen

Verteilen Sie Ihre Last auf mehrere Worker – CaptchaAI ausprobieren.