Kafka compensa a partir do momento em que a fila de tarefas CAPTCHA do seu scraper deixa de ser um detalhe e vira gargalo — na prática, quando o volume passa de alguns milhares de resoluções por hora e uma fila simples começa a acumular atraso. O Kafka separa quem envia o desafio de quem processa o token, grava as mensagens em disco (dá para reprocessar depois) e escala horizontalmente por partição. É essa combinação que sustenta pipelines de CAPTCHA de alto volume em produção.
Arquitetura do pipeline de streaming
[Scrapers] → Produce → [Kafka: captcha-tasks topic]
↓
[CAPTCHA Worker Group]
(consume tasks, solve via CaptchaAI)
↓
Produce → [Kafka: captcha-results topic]
↓
[Result Consumer Group]
(process solutions, update database)
Dois tópicos cuidam de partes diferentes do fluxo: captcha-tasks recebe os desafios aguardando resolução, e captcha-results guarda os tokens já resolvidos, prontos para o consumidor final usar. Separar os dois evita que quem envia tarefas precise saber qualquer coisa sobre quem consome resultados.
O que você precisa ter pronto
# Python
pip install kafka-python requests
# Node.js
npm install kafkajs axios
Além dos pacotes acima, você precisa de um broker Kafka rodando — localhost:9092 para testes locais, ou o endereço do seu cluster em produção.
Passo 1: crie os tópicos
kafka-topics.sh --create --topic captcha-tasks \
--partitions 6 --replication-factor 1 \
--bootstrap-server localhost:9092
kafka-topics.sh --create --topic captcha-results \
--partitions 6 --replication-factor 1 \
--bootstrap-server localhost:9092
Com seis partições, dá para rodar até seis consumers em paralelo dentro do mesmo grupo — é esse número que decide o teto de paralelismo antes de você precisar recriar o tópico.
Passo 2: o producer de tarefas (lado do scraper)
Em Python, o producer conecta no broker e serializa cada tarefa como JSON antes de publicar:
import json
from kafka import KafkaProducer
producer = KafkaProducer(
bootstrap_servers=["localhost:9092"],
value_serializer=lambda v: json.dumps(v).encode("utf-8"),
key_serializer=lambda k: k.encode("utf-8") if k else None,
acks="all", # Wait for all replicas to confirm
retries=3
)
def enqueue_captcha(task_id, sitekey, pageurl, captcha_type="userrecaptcha"):
"""Send a CAPTCHA task to Kafka."""
task = {
"task_id": task_id,
"method": captcha_type,
"sitekey": sitekey,
"pageurl": pageurl,
"submitted_at": __import__("time").time()
}
future = producer.send(
"captcha-tasks",
key=task_id, # Key ensures same task goes to same partition
value=task
)
future.get(timeout=10) # Block until confirmed
return task_id
# Submit tasks
enqueue_captcha("task_001", "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-", "https://example.com")
enqueue_captcha("task_002", "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-", "https://example.com")
producer.flush()
O equivalente em Node.js usa o kafkajs, com a mesma lógica de conectar, montar a tarefa e publicar no tópico:
const { Kafka } = require("kafkajs");
const kafka = new Kafka({
clientId: "captcha-producer",
brokers: ["localhost:9092"],
});
const producer = kafka.producer();
async function enqueueCaptcha(taskId, sitekey, pageurl) {
await producer.connect();
const task = {
task_id: taskId,
method: "userrecaptcha",
sitekey: sitekey,
pageurl: pageurl,
submitted_at: Date.now(),
};
await producer.send({
topic: "captcha-tasks",
messages: [{ key: taskId, value: JSON.stringify(task) }],
});
}
(async () => {
await enqueueCaptcha(
"task_001",
"6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
"https://example.com"
);
await producer.disconnect();
})();
Passo 3: o worker que resolve o CAPTCHA (consumer + solver)
O worker consome o tópico captcha-tasks, chama a API da CaptchaAI e publica o resultado em captcha-results. Em Python:
import json
import os
import time
import requests
from kafka import KafkaConsumer, KafkaProducer
API_KEY = os.environ["CAPTCHAAI_API_KEY"]
consumer = KafkaConsumer(
"captcha-tasks",
bootstrap_servers=["localhost:9092"],
group_id="captcha-workers",
value_deserializer=lambda m: json.loads(m.decode("utf-8")),
auto_offset_reset="earliest",
enable_auto_commit=False, # Manual commit after processing
max_poll_records=10
)
result_producer = KafkaProducer(
bootstrap_servers=["localhost:9092"],
value_serializer=lambda v: json.dumps(v).encode("utf-8")
)
def solve_captcha(task):
"""Submit to CaptchaAI and poll for result."""
# Submit
resp = requests.post("https://ocr.captchaai.com/in.php", data={
"key": API_KEY,
"method": task["method"],
"googlekey": task["sitekey"],
"pageurl": task["pageurl"],
"json": 1
})
data = resp.json()
if data.get("status") != 1:
return {"error": data.get("request")}
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 {"solution": result["request"]}
if result.get("request") != "CAPCHA_NOT_READY":
return {"error": result.get("request")}
return {"error": "TIMEOUT"}
# Main consumer loop
print("CAPTCHA worker started. Waiting for tasks...")
for message in consumer:
task = message.value
print(f"Processing {task['task_id']}...")
result = solve_captcha(task)
result["task_id"] = task["task_id"]
result["solved_at"] = time.time()
# Publish result
result_producer.send("captcha-results", value=result)
result_producer.flush()
# Commit offset after successful processing
consumer.commit()
print(f" → {task['task_id']}: {'solved' if 'solution' in result else result.get('error')}")
Em Node.js, a mesma lógica com kafkajs e axios:
const { Kafka } = require("kafkajs");
const axios = require("axios");
const API_KEY = process.env.CAPTCHAAI_API_KEY;
const kafka = new Kafka({
clientId: "captcha-worker",
brokers: ["localhost:9092"],
});
const consumer = kafka.consumer({ groupId: "captcha-workers" });
const producer = kafka.producer();
function sleep(ms) {
return new Promise((resolve) => setTimeout(resolve, ms));
}
async function solveCaptcha(task) {
const submitResp = await axios.post(
"https://ocr.captchaai.com/in.php",
null,
{
params: {
key: API_KEY,
method: task.method,
googlekey: task.sitekey,
pageurl: task.pageurl,
json: 1,
},
}
);
if (submitResp.data.status !== 1) {
return { error: submitResp.data.request };
}
const captchaId = submitResp.data.request;
for (let i = 0; i < 60; i++) {
await sleep(5000);
const result = await axios.get("https://ocr.captchaai.com/res.php", {
params: { key: API_KEY, action: "get", id: captchaId, json: 1 },
});
if (result.data.status === 1) return { solution: result.data.request };
if (result.data.request !== "CAPCHA_NOT_READY")
return { error: result.data.request };
}
return { error: "TIMEOUT" };
}
async function run() {
await consumer.connect();
await producer.connect();
await consumer.subscribe({ topic: "captcha-tasks", fromBeginning: false });
await consumer.run({
eachMessage: async ({ message }) => {
const task = JSON.parse(message.value.toString());
console.log(`Processing ${task.task_id}...`);
const result = await solveCaptcha(task);
result.task_id = task.task_id;
result.solved_at = Date.now();
await producer.send({
topic: "captcha-results",
messages: [{ value: JSON.stringify(result) }],
});
console.log(
` → ${task.task_id}: ${result.solution ? "solved" : result.error}`
);
},
});
}
run();
Validação de mensagens no producer
Antes de colocar isso em produção, vale travar três coisas no lado de quem publica:
- Rejeite mensagens com tipo de solver, metadados de destino ou detalhes de roteamento de resposta inválidos antes que elas cheguem ao tópico.
- Use uma chave de idempotência para que uma retentativa não crie uma segunda tentativa de resolução para o mesmo desafio.
- Envie registros malformados para uma dead-letter queue com contexto para reproduzir o problema depois. Se você registra
pageurlou outros metadados de tarefa nesses logs, trate como dado sujeito à LGPD e aplique retenção curta.
Problemas comuns e como resolver
- Lag do consumer crescendo: os workers não acompanham a taxa de tarefas — adicione mais instâncias de worker, até o número de partições.
- Resultados duplicados: o worker trava antes de confirmar o offset — adicione verificação de idempotência por
task_idno consumer de resultados. - Rebalanceamento frequente demais: workers travando ou reiniciando — aumente
session.timeout.mse verifique se há OOM. - Tarefas mal distribuídas entre partições: distribuição de chaves ruim — use chaves aleatórias ou crie mais partições.
Escalando os workers horizontalmente
# 6 partitions, 3 workers → each worker gets 2 partitions
Worker-1: partitions 0, 1
Worker-2: partitions 2, 3
Worker-3: partitions 4, 5
# Add Worker-4 → rebalance
Worker-1: partitions 0, 1
Worker-2: partitions 2
Worker-3: partitions 3, 4
Worker-4: partition 5
Os consumer groups do Kafka distribuem as partições entre os workers sozinhos — você não programa esse balanceamento. Dá para escalar até o número de partições que você criou; cada worker a mais reduz a fila de outro. Depois disso, a única saída é aumentar o número de partições e redistribuir o tráfego.
Monitorando o pipeline em produção
O lag do consumer group é a métrica mais importante para saber se o pipeline está acompanhando o ritmo de tarefas:
kafka-consumer-groups.sh --describe --group captcha-workers \
--bootstrap-server localhost:9092
- Lag do consumer: saudável abaixo de 100; acima de 1000, adicione workers.
- Mensagens/s recebidas: deve acompanhar a taxa do scraper — picos indicam rajada de tarefas.
- Mensagens/s publicadas: deve acompanhar a taxa de entrada — ficar para trás é sinal de gargalo no worker.
Scrapers no Brasil ganham com broker e workers na região sa-east-1 da AWS (São Paulo): RTT menor até a API da CaptchaAI ajuda a manter o polling dentro do timeout em picos de tráfego.
Perguntas frequentes
Preciso de Kafka desde o início do projeto?
Não. Kafka compensa a partir de alguns milhares de tarefas por hora, quando você precisa de replay e de vários workers em paralelo. Abaixo disso, Redis ou RabbitMQ resolvem com bem menos infraestrutura.
Dá para usar um Kafka gerenciado (Confluent Cloud, MSK) em vez de self-hosted?
Sim, e é a opção mais comum em produção — o código do producer e do consumer não muda, só os parâmetros de conexão (bootstrap_servers, credenciais SASL/TLS). O que muda é quem cuida do broker.
O que acontece se dois workers pegarem a mesma tarefa?
Não acontece dentro do mesmo consumer group — o Kafka garante que cada partição seja lida por um único worker por vez. Duplicatas só surgem se um worker travar depois de processar mas antes de confirmar o offset; por isso a checagem de idempotência por task_id é obrigatória.
Como sei se o pipeline está atrasado sem esperar reclamação do time?
Monitore o lag do consumer group (kafka-consumer-groups.sh --describe) e alerte acima de ~1000 mensagens de atraso. Lag crescendo de forma constante — não só em picos — é o sinal mais confiável de que faltam workers.
Artigos relacionados
- Como processar resultados de CAPTCHA em lote via streaming
Próximas etapas
Monte seu pipeline de CAPTCHA em streaming: crie sua chave de API da CaptchaAI e conecte o Kafka para sustentar alto volume sem perder tarefas.
Guias relacionados:
