asyncio 功能强大，但需要将整个调用链重写为异步。 ThreadPoolExecutor 为您提供与标准同步代码的并行性 - 将其放入现有项目中而无需重组。
为什么使用 ThreadPoolExecutor 进行验证码
验证码解析为I/O-bound（等待HTTP响应）。 Python 线程在 I/O 操作期间释放 GIL，使 ThreadPoolExecutor 对此工作负载高效：
| 方法 | 复杂 | 适合现有代码 | I/O 的并行性 |
|---|---|---|---|
| 顺序 | 没有任何 | 是的 | 没有任何 |
| 线程池执行器 | 低的 | 是的 | 好的 |
| 异步 | 高的 | 需要异步重写 | 最好的 |
| 多重处理 | 中等的 | 大多 | 对 I/O 来说太过分了 |
基本实现
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")
使用会话进行连接重用
每个请求创建一个新的 TCP 连接会浪费时间。每个线程共享一个 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")
map() 用于简单的批量操作
当您不需要每个任务的错误处理时：
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)}")
超时保护
防止失控线程阻塞池：
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}")
进度回调
实时跟踪完成情况：
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
选择 max_workers
| 工人 | 并发解决 | 开销 | 最适合 |
|---|---|---|---|
| 5 | 5 | 很低 | 小批量，保守使用 |
| 10 | 10 | 低的 | 一般用途 |
| 25 | 25 | 缓和 | 大容量管道 |
| 50 | 50 | 更高 | 最大吞吐量 |
更多的工作线程意味着更多的并发 API 连接。从 10 开始，在监控错误率的同时增加。
ThreadPoolExecutor 与 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)
在以下情况下使用 ThreadPoolExecutor：
- 您现有的代码库是同步的
- 您使用不支持异步的库（Selenium、某些 ORM）
- 您想要快速并行而不需要重组
在以下情况下使用 asyncio：
- 从头开始构建
- 最大效率很重要（更少的操作系统线程）
- 已经在异步框架中（FastAPI、aiohttp）
故障排除
| 问题 | 原因 | 处理方式 |
|---|---|---|
| 所有线程都被阻塞 | 每个线程在轮询期间等待 time.sleep |
这是预期的——线程在睡眠期间释放 GIL |
ConnectionError 秒杀 |
并发连接数过多 | 减少max_workers；使用连接池 |
| 结果乱序 | as_completed 按完成顺序返回 |
使用 map() 获取有序结果，或使用 dict 进行跟踪 |
| 记忆力增长 | 期货中持有的大型结果对象 | as_completed循环中处理结果；不存储全部 |
常问问题
GIL 会阻止真正的并行吗？
不 - 对于像 HTTP 请求和 time.sleep 这样的 I/O-bound 工作，Python 释放了 GIL。您的线程在网络调用期间真正并发运行。GIL 仅限制 CPU 绑定的并行性。
ThreadPoolExecutor 每小时可以处理多少个验证码？
10 名工作人员和 15 秒的平均解决时间：每小时约 2,400 个。拥有 25 名工人：每小时约 6,000 人。瓶颈是 CaptchaAI 求解时间，而不是 Python 线程。
我应该使用 ProcessPoolExecutor 吗？
不会。验证码解决是 I/O-bound. ProcessPoolExecutor 增加了进程间通信开销，但没有任何好处。坚持使用线程。
下一步
并行验证码解决 -获取您的 CaptchaAI API 密钥并将 ThreadPoolExecutor 放入管道中。
相关指南：
