Se o seu script resolve CAPTCHAs um de cada vez, o gargalo não é a CaptchaAI — é o time.sleep esperando cada resposta antes da próxima. A correção não é migrar tudo para asyncio: é trocar o laço sequencial por um ThreadPoolExecutor, sobre o código síncrono que você já tem.
Por que o ThreadPoolExecutor é a escolha certa para CAPTCHA
Resolver CAPTCHA é I/O-bound — o tempo é gasto esperando respostas HTTP, não processando dados. Threads em Python liberam o GIL durante I/O, então o ThreadPoolExecutor entrega paralelismo real sem a complexidade de reescrever o projeto em async:
- Sequencial — zero complexidade, nenhum paralelismo.
- ThreadPoolExecutor — baixa complexidade, encaixa em código síncrono, bom paralelismo para I/O.
- asyncio — exige reescrever a cadeia de chamadas em async; paralelismo máximo.
- multiprocessing — overhead desnecessário para uma carga que é só espera de rede.
Implementação básica: enviar e consultar em paralelo
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")
O que muda em relação a um laço for comum
solve_captcha é uma função síncrona comum — sem async def, sem await. Quem orquestra as 20 tarefas em 10 threads é o with ThreadPoolExecutor(...).
Por que enviar tudo antes de ler os resultados
executor.submit(...) dispara as 20 chamadas de imediato — é o que gera o paralelismo. Só depois o segundo laço lê cada future conforme termina.
Sessões por thread para reaproveitar conexões
Abrir uma conexão TCP nova a cada requisição desperdiça tempo com handshake. Compartilhe um requests.Session por thread — não uma sessão global, nem uma nova por chamada:
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")
pool_maxsize deve acompanhar max_workers, ou você cria fila interna e perde parte do ganho.
map() para lotes simples, sem tratar erro tarefa a tarefa
Sem lógica de retentativa individual, executor.map() é mais direto que 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)}")
Protegendo o pool contra tempo limite
Sem limite explícito, uma thread travada pode segurar o pool inteiro. Combine timeout global e por tarefa:
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}")
Acompanhando o progresso em tempo real
Em pipelines com dezenas de tarefas, um contador evita a sensação de "travou":
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
Quantos workers usar
Depende do plano contratado e do volume da fila:
- 5 workers — overhead muito baixo; lotes pequenos, uso conservador.
- 10 workers — overhead baixo; bom padrão geral.
- 25 workers — overhead moderado; pipelines de alto volume.
- 50 workers — overhead maior; só compensa em throughput máximo.
Mais workers = mais conexões simultâneas — cada plano define quantas threads você tem (BASIC inclui 5 threads por US$ 15/mês). Comece em max_workers=10 e suba conforme o plano permitir.
ThreadPoolExecutor ou asyncio: qual escolher
# 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)
- Use
ThreadPoolExecutorcom base síncrona, libs sem async (Selenium, alguns ORMs) ou para paralelismo rápido sem reestruturar. - Use
asyncioem projetos novos, quando eficiência máxima importa, ou já dentro de um framework async (FastAPI, aiohttp).
E se os workers rodarem fora dos EUA?
Rodando de uma região como AWS sa-east-1 (São Paulo), a latência até o endpoint da CaptchaAI se soma ao tempo total — mas não muda o cálculo de max_workers, já que a thread libera o GIL nessa espera também. Meça o RTT antes de fixar um timeout agressivo por tarefa.
Problemas comuns e como resolver
- Threads parecem travadas — cada uma está em
time.sleepaguardando o polling; é esperado, o GIL é liberado durante o sleep. - Picos de
ConnectionError— conexões simultâneas demais para o pool; reduzamax_workersou aumentepool_maxsize. - Resultados fora de ordem —
as_completedretorna por conclusão, não por envio; usemap()para ordem preservada. - Memória crescendo — resultados grandes acumulados nos futures; processe cada item dentro do laço
as_completed.
Perguntas frequentes
O GIL realmente permite paralelismo aqui, ou é ilusão?
É real. Durante chamadas de rede e time.sleep, o Python libera o GIL, e as threads avançam ao mesmo tempo. O GIL só vira gargalo em trabalho vinculado a CPU — não é o caso aqui.
Quantas threads devo configurar no ThreadPoolExecutor?
Comece em max_workers=10 e ajuste conforme o plano e a taxa de erro. Configurar mais workers do que threads o plano permite não ajuda — o excesso só fica na fila.
O ThreadPoolExecutor funciona com Selenium e outras libs sem suporte a async?
Sim — é a vantagem sobre asyncio. Sem API assíncrona nativa no Selenium, rodar várias instâncias dentro de um ThreadPoolExecutor paraleliza automação de navegador junto com a resolução de CAPTCHA.
Quantos CAPTCHAs dá para resolver por hora com esse padrão?
Com 10 workers e 15 segundos de tempo médio de resolução: cerca de 2.400 por hora. Com 25 workers: cerca de 6.000 por hora. O limite é o tempo de resolução por tipo de desafio, não o threading do Python.
Como começar agora
Pare de resolver CAPTCHA um por vez — obtenha sua chave de API da CaptchaAI e coloque o ThreadPoolExecutor no seu pipeline hoje mesmo.
Guias relacionados:
