Python ThreadPoolExecutor để giải quyết song song CAPTCHA

asyncio mạnh mẽ nhưng yêu cầu viết lại toàn bộ chuỗi cuộc gọi của bạn dưới dạng không đồng bộ. ThreadPoolExecutor cung cấp cho bạn khả năng song song với mã đồng bộ tiêu chuẩn - thả nó vào các dự án hiện có mà không cần cơ cấu lại.

Tại sao ThreadPoolExecutor dành cho CAPTCHA

Giải CAPTCHA là I/O-bound (chờ phản hồi HTTP). Các luồng Python giải phóng GIL trong các hoạt động I/O, giúp ThreadPoolExecutor hoạt động hiệu quả cho khối lượng công việc này:

Triển khai cơ bản

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

Sử dụng phiên để tái sử dụng kết nối

Tạo kết nối TCP mới cho mỗi yêu cầu sẽ lãng phí thời gian. Chia sẻ requests.Session trên mỗi chủ đề:

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() cho các hoạt động hàng loạt đơn giản

Khi bạn không cần xử lý lỗi theo từng tác vụ:

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

Bảo vệ thời gian chờ

Ngăn chặn các chủ đề chạy trốn chặn nhóm của bạn:

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

Gọi lại tiến độ

Theo dõi hoàn thành trong thời gian thực:

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

Chọn max_workers

Nhiều công nhân hơn có nghĩa là có nhiều kết nối API đồng thời hơn. Bắt đầu ở mức 10, tăng dần trong khi theo dõi tỷ lệ lỗi.

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

Sử dụng ThreadPoolExecutor khi:

• Cơ sở mã hiện tại của bạn là đồng bộ
• Bạn sử dụng các thư viện không hỗ trợ async (Selenium, một số ORM)
• Bạn muốn song song nhanh chóng mà không cần tái cơ cấu

Sử dụng asyncio khi:

• Xây dựng từ đầu
• Vấn đề hiệu quả tối đa (ít luồng hệ điều hành hơn)
• Đã có trong khung không đồng bộ (FastAPI, aiohttp)

Khắc phục sự cố

Câu hỏi thường gặp

GIL có ngăn chặn sự song song thực sự không?

Không - để I/O-bound hoạt động giống như các yêu cầu HTTP và time.sleep, Python sẽ phát hành GIL. Chủ đề của bạn chạy thực sự đồng thời trong các cuộc gọi mạng. GIL chỉ giới hạn tính song song của CPU.

ThreadPoolExecutor có thể xử lý bao nhiêu CAPTCHA mỗi giờ?

Với 10 công nhân và thời gian giải quyết trung bình 15 giây: ~2.400 mỗi giờ. Với 25 công nhân: ~6.000/giờ. Nút thắt là thời gian giải quyết CaptchaAI chứ không phải luồng Python.

Tôi có nên sử dụng ProcessPoolExecutor thay thế không?

Không. Việc giải CAPTCHA là I/O-bound. ProcessPoolExecutor bổ sung chi phí liên lạc giữa các quá trình mà không mang lại lợi ích gì. Gắn bó với chủ đề.

Các bước tiếp theo

Song song hóa việc giải CAPTCHA –lấy khóa API CaptchaAI của bạnvà thả ThreadPoolExecutor vào đường dẫn của bạn.

Hướng dẫn liên quan:

• Giải quyết CAPTCHA song song
• Hiệu suất song song và tuần tự
• Giải quyết 10.000 nhiệm vụ mỗi giờ