Python ThreadPoolExecutor untuk Paralelisme Solve CAPTCHA

Dengan 10 worker dan waktu solve rata-rata 15 detik, ThreadPoolExecutor menuntaskan sekitar 2.400 CAPTCHA per jam — tanpa satu baris async/await. Itu jawabannya. Kalau codebase Anda sudah sinkron (requests, Selenium, Flask), ThreadPoolExecutor dari concurrent.futures adalah cara paling praktis menambah paralelisme tanpa membongkar arsitektur yang ada, beda dari asyncio yang menuntut seluruh rantai pemanggilan ditulis ulang jadi async.

Ini relevan buat tim scraping/automation Indonesia yang cost-sensitive — freelancer Upwork/Fastwork maupun tim data startup — karena paralelisme masuk ke kode sinkron yang sudah jalan, bukan lewat rewrite dari nol.

Implementasi Dasar: Submit dan Polling Paralel

Mulai dari kode yang langsung bisa dicopy: fungsi solve sinkron biasa — submit ke in.php, polling res.php — dijalankan lewat pool 10 worker.

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

Kenapa Ini Bekerja: I/O-Bound dan GIL

Solve CAPTCHA itu I/O-bound: waktu terbanyak habis menunggu respons in.php/res.php, bukan menghitung. Saat menunggu I/O, thread Python melepaskan GIL (Global Interpreter Lock) sehingga thread lain tetap jalan bersamaan. Itu sebabnya pool sederhana di atas sudah cukup, dan kenapa pendekatan lain sering berlebihan untuk kasus ini:

• Berurutan (tanpa pool) — kompleksitas nol, tapi nol paralelisme; setiap solve menunggu solve sebelumnya selesai.
• ThreadPoolExecutor — kompleksitas rendah, langsung cocok dengan kode existing, paralelisme I/O bagus.
• asyncio — paralelisme I/O terbaik secara teori, tapi kompleksitas tinggi karena seluruh rantai panggilan harus ditulis ulang jadi async.
• multiprocessing — kompleksitas sedang, overhead antar-proses yang berlebihan untuk beban kerja yang sebetulnya cuma menunggu jaringan.

Reuse Koneksi dengan Session per Thread

Koneksi TCP baru per request itu boros. Solusinya: satu requests.Session per thread, disimpan di threading.local(), dengan connection pooling lewat HTTPAdapter:

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() untuk Batch Sederhana Tanpa Penanganan Error Manual

Tidak perlu penanganan error per task? executor.map() lebih ringkas — tiap task mengembalikan dict berisi hasil atau error, lalu difilter setelahnya:

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

Cegah Thread Nyangkut dengan Timeout Ganda

Satu thread yang macet bisa mengunci seluruh pool. Pasang dua lapis timeout: timeout global di as_completed() dan timeout per task di future.result():

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

Lacak Progress Secara Real-Time

Untuk batch besar, tampilkan progress berjalan pakai threading.Lock supaya counter tidak race antar thread:

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

Memilih max_workers yang Tepat untuk Traffic Anda

Berapa banyak worker tergantung volume dan toleransi risiko Anda:

Lebih banyak worker = lebih banyak koneksi API bersamaan. Mulai dari 10, naikkan bertahap sambil memantau error rate, bukan cuma throughput mentah — pada 25 worker angkanya mendekati 6.000 CAPTCHA/jam, tapi bottleneck-nya waktu solve CaptchaAI, bukan threading Python. Menambah worker di atas titik itu cuma menumpuk koneksi menganggur.

ThreadPoolExecutor vs asyncio: Kapan Pakai yang Mana

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

Kapan pakai ThreadPoolExecutor

• Basis kode Anda sudah sinkron
• Anda memakai library tanpa dukungan async (Selenium, sebagian ORM)
• Butuh paralelisme cepat tanpa membongkar arsitektur

Kapan pakai asyncio

• Membangun dari nol
• Efisiensi maksimum penting (thread OS lebih sedikit)
• Sudah di framework async (FastAPI, aiohttp)

Soal ProcessPoolExecutor — sebaiknya jangan dipakai di sini. Solve CAPTCHA itu I/O-bound, jadi overhead komunikasi antar-proses cuma menambah beban tanpa manfaat nyata dibanding thread biasa.

Troubleshooting Masalah Umum

Semua thread tampak diam

Ini normal. Tiap thread sedang menunggu time.sleep saat polling — GIL dilepas selama sleep, jadi thread lain tetap jalan di belakang layar.

Lonjakan ConnectionError

Terlalu banyak koneksi bersamaan. Turunkan max_workers, dan pastikan connection pooling lewat HTTPAdapter sudah aktif.

Hasil keluar tidak berurutan

as_completed mengembalikan future begitu selesai, bukan menurut urutan submit. Pakai map() kalau butuh hasil terurut, atau lacak lewat dict seperti pada contoh di atas.

Memori terus naik saat batch besar

Objek hasil besar tertahan di future sebelum diproses. Proses tiap hasil langsung di loop as_completed, jangan simpan semuanya sekaligus di list.

Pertanyaan yang Sering Diajukan

Apakah ThreadPoolExecutor bisa langsung dipasang ke project Selenium atau Flask yang sudah jalan?

Bisa. Kode sinkron biasa — tidak perlu menulis ulang fungsi jadi async def atau mengganti stack yang sudah dipakai. Bungkus fungsi solve CAPTCHA yang ada, submit ke pool.

Apakah GIL Python jadi penghambat saat solve CAPTCHA secara paralel?

Tidak. Solve CAPTCHA itu I/O-bound — menunggu HTTP dan time.sleep saat polling. Python melepaskan GIL selama I/O, jadi thread benar-benar berjalan bersamaan. GIL hanya membatasi kerja CPU-bound.

Berapa max_workers yang aman untuk koneksi kantor atau mobile yang naik-turun?

Mulai dari 5–10, terutama di jaringan mobile atau koneksi yang kurang stabil. Timeout ganda (per task dan global) jadi lebih penting di kondisi ini — task yang macet karena koneksi putus tidak akan mengunci seluruh pool.

Kalau saya naikkan max_workers, apakah otomatis butuh paket thread CaptchaAI yang lebih besar?

Ya. Worker paralel di kode Anda dibatasi kuota thread paket CaptchaAI — BASIC ($15/bulan) mencakup 5 thread, ADVANCE ($90/bulan) mencakup 50 thread. Set max_workers sesuai kuota; melebihi jumlah thread paket hanya membuat request mengantre, bukan mempercepat throughput.

Apakah lokasi server worker (Singapura vs Jakarta) memengaruhi throughput?

Sedikit. Latency ke CaptchaAI dari region seperti AWS ap-southeast-1 (Singapura) atau GCP asia-southeast2 (Jakarta) lebih rendah dibanding server di luar Asia Tenggara, jadi setiap siklus polling sedikit lebih cepat. Efeknya kecil dibanding waktu solve itu sendiri, tapi lumayan untuk pipeline bervolume tinggi.

Langkah Selanjutnya

Paralelkan solve CAPTCHA Anda —dapatkan API key CaptchaAIlalu pasang ThreadPoolExecutor ke pipeline yang ada.

Baca juga:

• Cara Kerja Solve CAPTCHA Paralel
• Paralel vs Berurutan: Mana yang Lebih Cepat
• Cara Menuntaskan 10.000 Task CAPTCHA per Jam