Một pipeline giải tuần tự mất khoảng 4 giờ để xử lý 1.000 CAPTCHA. Chuyển sang ThreadPoolExecutor với 10 worker, thời gian rút còn khoảng 25 phút — không cần viết lại dòng nào sang asyncio. Mã đồng bộ giữ nguyên, chỉ thêm một lớp thread pool.
Vì sao ThreadPoolExecutor hợp với bài toán giải CAPTCHA
Giải CAPTCHA là tác vụ I/O-bound — phần lớn thời gian chờ phản hồi HTTP. Python nhả GIL trong lúc chờ I/O, nên nhiều thread chạy thật sự song song khi gọi mạng. So các cách tiếp cận:
- Tuần tự — không phức tạp, nhưng không song song I/O: request sau chờ request trước xong.
- ThreadPoolExecutor — độ phức tạp thấp, chạy thẳng trên code đồng bộ hiện có.
- asyncio — độ phức tạp cao, buộc viết lại thành async, song song I/O tốt nhất.
- multiprocessing — độ phức tạp trung bình, nhưng dư thừa cho tác vụ I/O-bound như giải CAPTCHA.
Cách triển khai cơ bản
Hàm dưới đây gửi task tới in.php, sau đó polling res.php cho tới khi có token — logic đồng bộ y hệt gọi API tuần tự, chỉ khác ở chỗ chạy trong thread pool:
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")
Dùng chung session để tái sử dụng kết nối
Mở kết nối TCP mới cho từng request là lãng phí. Mỗi thread nên giữ riêng một requests.Session:
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")
Giữ pool_maxsize khớp với max_workers — lệch hai số là nguyên nhân phổ biến nhất của log "Connection pool is full".
map() khi không cần xử lý lỗi riêng từng task
Chỉ cần kết quả theo đúng thứ tự input? executor.map() gọn hơn as_completed:
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)}")
Chặn timeout để tránh treo cả pool
Một thread bị treo không được phép giữ cả pool chờ vô thời hạn. Đặt timeout ở cả hai cấp — toàn cục và từng task:
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}")
Theo dõi tiến độ bằng callback
In tiến độ theo thời gian thực giúp phát hiện sớm nếu tỷ lệ lỗi tăng bất thường:
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
progress_lock bắt buộc phải có — nhiều thread cùng cộng vào progress["done"] mà không khóa sẽ ra số đếm sai.
Chọn max_workers bao nhiêu là hợp lý
Càng nhiều worker, càng nhiều kết nối API mở đồng thời — không phải cứ đặt số lớn là nhanh hơn tuyến tính:
- 5 worker — chi phí rất thấp, hợp lô nhỏ hoặc chạy thận trọng.
- 10 worker — chi phí thấp, điểm khởi đầu hợp lý cho phần lớn pipeline.
- 25 worker — chi phí trung bình, phù hợp pipeline khối lượng lớn.
- 50 worker — chi phí cao hơn, dùng khi cần thông lượng tối đa.
Bắt đầu ở mức 10, tăng dần theo tỷ lệ lỗi — đừng nhảy thẳng lên 50 chỉ vì máy chủ rảnh.
Lưu ý:
max_workerslà thread phía client, khác với "thread" trong bảng giá CaptchaAI (luồng giải đồng thời phía server). Ví dụ: đội QA outsourcing tại TP.HCM kiểm thử luồng đăng ký sàn thương mại điện tử, cần giải vài nghìn reCAPTCHA v2/ngày trên staging. Đặtmax_workers=10nhưng tài khoản ở gói BASIC ($15/tháng, 5 thread) thì chỉ 5 request chạy cùng lúc — phần còn lại xếp hàng; gói STANDARD ($30/tháng, 15 thread) mới đủ thread để khớp.
ThreadPoolExecutor hay 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)
Chọn ThreadPoolExecutor khi:
- Codebase hiện tại viết theo kiểu đồng bộ
- Bạn phụ thuộc thư viện chưa hỗ trợ async (Selenium, một số ORM)
- Cần song song hóa nhanh mà không tái cấu trúc dự án
Chọn asyncio khi:
- Xây dựng pipeline mới hoàn toàn từ đầu
- Hiệu năng tối đa là ưu tiên số một
- Codebase đã nằm trong framework async sẵn (FastAPI, aiohttp)
Các lỗi thường gặp khi dùng ThreadPoolExecutor
Toàn bộ thread bị treo
Bình thường — mọi thread đang chờ ở time.sleep trong lúc polling, đó là lúc GIL được nhả, không phải bug.
ConnectionError tăng đột biến
Quá nhiều kết nối đồng thời. Giảm max_workers hoặc bật connection pooling như ở phần session dùng chung.
Kết quả trả về không đúng thứ tự
as_completed trả theo thứ tự hoàn thành, không theo thứ tự gửi. Dùng map() để giữ đúng thứ tự đầu vào.
Bộ nhớ tăng dần trong lúc chạy
Future giữ kết quả lớn trong bộ nhớ tới khi bạn đọc. Xử lý ngay trong vòng lặp as_completed thay vì gom hết vào một list.
Câu hỏi thường gặp
Chạy song song có tốn thêm phí CaptchaAI không?
Không. CaptchaAI tính phí theo thread cố định hằng tháng, không theo số lần giải nên song song hoá không tốn thêm nếu vẫn trong hạn mức thread gói.
Nên đặt max_workers bao nhiêu cho pipeline giải CAPTCHA?
Bắt đầu ở 10 và khớp với số thread của gói CaptchaAI đang dùng — đặt cao hơn chỉ khiến request xếp hàng. Khối lượng lớn nên dùng gói CORPORATE ($240/tháng, 150 thread) hoặc ENTERPRISE ($300/tháng, 200 thread).
ThreadPoolExecutor giải được bao nhiêu CAPTCHA mỗi giờ?
Với 10 worker, thời gian giải trung bình 15 giây: khoảng 2.400 CAPTCHA/giờ. Nút thắt là thời gian giải của CaptchaAI, không phải số thread Python.
Có nên dùng ProcessPoolExecutor thay cho thread không?
Không. Giải CAPTCHA là I/O-bound; ProcessPoolExecutor sinh ra cho tác vụ tính toán nặng CPU và chỉ thêm chi phí liên tiến trình mà không tăng tốc độ giải.
Bước tiếp theo
Lấy API key CaptchaAI và thả ThreadPoolExecutor vào pipeline hiện có — không cần viết lại sang async.
Hướng dẫn liên quan:
