Tutoriels

Python ThreadPoolExecutor pour CAPTCHA résolvant le parallélisme

Automatisation Tous les types de CAPTCHA Python Performance
Par Équipe CaptchaAI France
Publié à Jul 16, 2026
Python ThreadPoolExecutor pour CAPTCHA résolvant le parallélisme
Image sélectionnée indisponible

asyncio est puissant mais nécessite de réécrire toute votre chaîne d'appels en asynchrone. ThreadPoolExecutor vous offre le parallélisme avec le code synchrone standard : déposez-le dans des projets existants sans restructuration.

Pourquoi ThreadPoolExecutor pour les CAPTCHA

La résolution de CAPTCHA est I/O-bound (en attente de réponses HTTP). Les threads Python libèrent le GIL pendant les opérations I/O, ce qui rend ThreadPoolExecutor efficace pour cette charge de travail :

Approche Complexité Correspond au code existant Parallélisme pour I/O
Séquentiel Aucun Oui Aucun
ThreadPoolExécuteur Faible Oui Bon
asyncio Élevé Nécessite une réécriture asynchrone Meilleur
multitraitement Moyen Surtout Exagération pour I/O

Implémentation de base

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

Utilisation de la session pour la réutilisation des connexions

Créer une nouvelle connexion TCP par requête fait perdre du temps. Partagez un requests.Session par fil :

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

map() pour les opérations par lots simples

Lorsque vous n'avez pas besoin de la gestion des erreurs par tâche :

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

Protection contre l'expiration du délai

Empêchez les threads incontrôlables de bloquer votre pool :

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

Rappel de progression

Suivez l'achèvement en temps réel :

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

Choisir max_workers

Travailleurs Résolutions simultanées Frais généraux Idéal pour
5 5 Très faible Petits lots, utilisation conservatrice
10 10 Faible Usage général
25 25 Modéré Pipelines à grand volume
50 50 Plus haut Débit maximal

Plus de travailleurs signifie plus de connexions API simultanées. Commencez à 10, augmentez tout en surveillant les taux d’erreur.

ThreadPoolExecutor contre 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)

Utilisez ThreadPoolExecutor lorsque :

  • Votre base de code existante est synchrone
  • Vous utilisez des bibliothèques qui ne prennent pas en charge l'async (Selenium, certains ORM)
  • Vous souhaitez un parallélisme rapide sans restructuration

Utilisez asyncio lorsque :

  • Construire à partir de zéro
  • L'efficacité maximale compte (moins de threads du système d'exploitation)
  • Déjà dans un framework asynchrone (FastAPI, aiohttp)

Dépannage

Problème Parce que Corriger
Tous les sujets bloqués Chaque thread en attente sur time.sleep pendant l'interrogation C'est normal : les threads libèrent GIL pendant le sommeil.
Pointes ConnectionError Trop de connexions simultanées Réduisez max_workers ; utiliser le regroupement de connexions
Des résultats dans le désordre as_completed revient dans l'ordre d'achèvement Utilisez map() pour les résultats ordonnés ou suivez avec dict
Mémoire croissante Objets à gros résultats détenus dans des contrats à terme Le processus aboutit à la boucle as_completed ; ne stocke pas tout

FAQ

Le GIL empêche-t-il un véritable parallélisme ?

Non – pour I/O-bound fonctionnant comme les requêtes HTTP et time.sleep, Python publie le GIL. Vos threads s'exécutent véritablement simultanément lors des appels réseau. Le GIL limite uniquement le parallélisme lié au processeur.

Combien de CAPTCHA ThreadPoolExecutor peut-il gérer par heure ?

Avec 10 travailleurs et un temps de résolution moyen de 15 secondes : ~2 400 par heure. Avec 25 ouvriers : ~6 000 par heure. Le goulot d'étranglement est le temps de résolution de CaptchaAI, et non le thread Python.

Dois-je plutôt utiliser ProcessPoolExecutor ?

Non. La résolution de CAPTCHA est I/O-bound.. ProcessPoolExecutor ajoute une surcharge de communication inter-processus sans aucun avantage. Tenez-vous-en aux fils.

Prochaines étapes

Paralléliser la résolution de CAPTCHA –récupérez votre clé API CaptchaAIet déposez ThreadPoolExecutor dans votre pipeline.

Guides associés :

Les commentaires sont désactivés pour cet article.