NATS Mesajlaşma + CaptchaAI: Basit CAPTCHA Görev Dağıtımı

Aynı anda yüzlerce CAPTCHA görevini çözmeniz gerektiğinde tek bir script yetmez; işi birden fazla worker'a dağıtmanız gerekir. NATS bunu en yalın haliyle yapar: tek bir ikili dosya, sıfır harici bağımlılık ve milisaniyenin altında gecikmeyle görevleri kuyruk gruplarına dağıtır. Kafka'nın disk dayanıklılığına ihtiyaç duymadığınız, hız ile sadeliğin öncelikli olduğu CAPTCHA iş yüklerinde NATS doğal seçimdir. Bu rehberde bir yayımlayıcı, kuyruk grubu tabanlı worker'lar ve bir sonuç toplayıcıyla CaptchaAI çözümlerini uçtan uca dağıtan çalışan bir kurulum kuracaksınız.

CAPTCHA görevleri için neden NATS?

CAPTCHA görevleri doğası gereği geçicidir: bir görev kaybolursa yeniden gönderirsiniz. Bu geçici iş yükünde NATS'ı öne çıkaran üç somut nokta şunlardır:

• Tek ikili, sıfır bağımlılık: ZooKeeper ya da ayrı bir küme yönetmeden dakikalar içinde ayağa kaldırırsınız.
• Milisaniyenin altında gecikme: darboğaz worker dağıtımı değil, yalnızca CaptchaAI çözüm süresi olur.
• İsteğe bağlı kalıcılık: varsayılan çekirdek mod hafiftir; kritik görevler için JetStream'i açarsınız.

Ne zaman NATS, ne zaman Kafka veya Redis?

Üç seçenek de dağıtık worker'ları besler, ancak karar iş yükünün doğasına bağlıdır:

• NATS: minimum altyapı, düşük gecikme, geçici görevler — çoğu CAPTCHA çözüm hattı buraya girer.
• Kafka: denetim izi, yeniden oynatma ve dayanıklı akış gerektiğinde.
• Redis: zaten önbellek olarak Redis kullanıyorsanız ve pub/sub'ı yanına eklemek istiyorsanız.

Yerel bir senaryo: e-ticaret QA ekibi

İstanbul'da bir e-ticaret ekibinin ödeme akışını QA testlerinde otomatikleştirdiğini düşünün: onlarca staging sayfasında reCAPTCHA v2 doğrulamasının çözülmesi gerekir. Kazıdığınız ya da test ettiğiniz veriler kişisel veri içeriyorsa bunun KVKK kapsamına girdiğini unutmayın; CaptchaAI'yi yetkili QA ve veri toplama akışları içinde kullanın. Maliyet tarafında CaptchaAI thread bazlı ücretlendirir; TL'nin oynaklığına karşı öngörülebilir aylık USD fiyatlar somut bir avantajdır. Örneğin ADVANCE plan ($90/ay, 50 thread) aynı anda 50 CAPTCHA çözmenize izin verir; worker sayınızı bu thread sınırına göre ölçekleyin.

Mimari: yayımlayıcı, worker'lar ve sonuç akışı

[Scrapers] → Publish → [NATS: captcha.tasks]
                              ↓
                    Queue Group: captcha-workers
                    ├── Worker 1 (solve via CaptchaAI)
                    ├── Worker 2
                    └── Worker 3
                              ↓
                    Publish → [NATS: captcha.results]
                              ↓
                    [Result Subscribers]

NATS kuyruk grupları mesajları worker'lar arasında otomatik olarak dengeler; her görev yalnızca bir worker'a ulaşır. Akış üç konu üzerinden ilerler:

• captcha.tasks — kazıyıcıların görevleri yayımladığı kuyruk.
• captcha-workers — her mesajı tek bir worker'a yönlendiren kuyruk grubu.
• captcha.results — çözülen ve başarısız görevlerin geri yazıldığı konu.

Ön koşullar: NATS sunucusu ve istemciler

Tek bir ikili dosyayla NATS sunucusunu kurun, ardından yayımlayıcı ve worker için istemciyi ekleyin. Kurulum üç parçadan oluşur:

• NATS sunucusu — tek ikili dosya, harici bağımlılık yok.
• İstemci kütüphanesi — Python için nats-py, Node.js için nats.
• CaptchaAI API anahtarı — worker'lara CAPTCHAAI_API_KEY ortam değişkeni olarak verilir.

# Install NATS server
# macOS
brew install nats-server

# Linux
curl -L https://github.com/nats-io/nats-server/releases/download/v2.10.0/nats-server-v2.10.0-linux-amd64.tar.gz | tar xz

# Start
nats-server

# Python client
pip install nats-py

# Node.js client
npm install nats

Görev yayımlayıcı (kazıyıcı tarafı)

Yayımlayıcı her CAPTCHA görevini JSON olarak captcha.tasks konusuna gönderir. Aşağıdaki örnek 100 görev üretir; sitekey ve pageurl değerlerini kendi hedefinizle değiştirin.

Python

import asyncio
import json
import nats


async def publish_captcha_tasks():
    nc = await nats.connect("nats://localhost:4222")

    tasks = [
        {
            "task_id": f"task_{i}",
            "method": "userrecaptcha",
            "sitekey": "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
            "pageurl": f"https://example.com/page/{i}"
        }
        for i in range(100)
    ]

    for task in tasks:
        await nc.publish("captcha.tasks", json.dumps(task).encode())
        print(f"Published: {task['task_id']}")

    await nc.flush()
    await nc.close()


asyncio.run(publish_captcha_tasks())

JavaScript

const { connect, StringCodec } = require("nats");

const sc = StringCodec();

async function publishCaptchaTasks() {
  const nc = await connect({ servers: "nats://localhost:4222" });

  for (let i = 0; i < 100; i++) {
    const task = {
      task_id: `task_${i}`,
      method: "userrecaptcha",
      sitekey: "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
      pageurl: `https://example.com/page/${i}`,
    };

    nc.publish("captcha.tasks", sc.encode(JSON.stringify(task)));
    console.log(`Published: ${task.task_id}`);
  }

  await nc.flush();
  await nc.close();
}

publishCaptchaTasks();

CAPTCHA worker'ı (kuyruk grubu abonesi)

Kuyruk grupları, birden fazla worker aynı anda çalışsa bile her mesajın yalnızca bir worker tarafından işlenmesini sağlar. Her worker aynı dört adımı izler:

1. Görevi captcha.tasks kuyruğundan alır.
2. CaptchaAI'ye gönderir (in.php).
3. Sonucu res.php ile sorgular.
4. Yanıtı captcha.results konusuna yazar.

Python

import asyncio
import json
import os
import nats
import aiohttp

API_KEY = os.environ["CAPTCHAAI_API_KEY"]


async def solve_captcha(session, task):
    """Submit to CaptchaAI and poll for result."""
    # Submit
    async with session.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": task["method"],
        "googlekey": task["sitekey"],
        "pageurl": task["pageurl"],
        "json": 1
    }) as resp:
        data = await resp.json(content_type=None)

    if data.get("status") != 1:
        return {"task_id": task["task_id"], "error": data.get("request")}

    captcha_id = data["request"]

    # Poll for result
    for _ in range(60):
        await asyncio.sleep(5)
        async with session.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY, "action": "get", "id": captcha_id, "json": 1
        }) as resp:
            result = await resp.json(content_type=None)

        if result.get("status") == 1:
            return {"task_id": task["task_id"], "solution": result["request"]}
        if result.get("request") != "CAPCHA_NOT_READY":
            return {"task_id": task["task_id"], "error": result.get("request")}

    return {"task_id": task["task_id"], "error": "TIMEOUT"}


async def worker(worker_id):
    nc = await nats.connect("nats://localhost:4222")

    # Subscribe with queue group — each message goes to one worker only
    sub = await nc.subscribe("captcha.tasks", queue="captcha-workers")

    print(f"Worker {worker_id} listening...")

    async with aiohttp.ClientSession() as session:
        async for msg in sub.messages:
            task = json.loads(msg.data.decode())
            print(f"Worker {worker_id} processing {task['task_id']}")

            result = await solve_captcha(session, task)

            # Publish result
            await nc.publish(
                "captcha.results",
                json.dumps(result).encode()
            )

            status = "solved" if "solution" in result else result.get("error")
            print(f"  → {task['task_id']}: {status}")


asyncio.run(worker(1))

JavaScript

const { connect, StringCodec } = require("nats");
const axios = require("axios");

const sc = StringCodec();
const API_KEY = process.env.CAPTCHAAI_API_KEY;

function sleep(ms) {
  return new Promise((r) => setTimeout(r, 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 { task_id: task.task_id, 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 { task_id: task.task_id, solution: result.data.request };
    }
    if (result.data.request !== "CAPCHA_NOT_READY") {
      return { task_id: task.task_id, error: result.data.request };
    }
  }

  return { task_id: task.task_id, error: "TIMEOUT" };
}

async function worker(workerId) {
  const nc = await connect({ servers: "nats://localhost:4222" });

  // Queue group subscription — load-balanced across workers
  const sub = nc.subscribe("captcha.tasks", { queue: "captcha-workers" });

  console.log(`Worker ${workerId} listening...`);

  for await (const msg of sub) {
    const task = JSON.parse(sc.decode(msg.data));
    console.log(`Worker ${workerId} processing ${task.task_id}`);

    const result = await solveCaptcha(task);
    nc.publish("captcha.results", sc.encode(JSON.stringify(result)));

    const status = result.solution ? "solved" : result.error;
    console.log(`  → ${task.task_id}: ${status}`);
  }
}

worker(1);

Sonuçları toplama

Sonuç toplayıcı captcha.results konusunu dinler ve çözülen ile başarısız görevleri sayarak canlı bir özet çıkarır.

async def collect_results():
    nc = await nats.connect("nats://localhost:4222")
    sub = await nc.subscribe("captcha.results")

    solved = 0
    failed = 0

    async for msg in sub.messages:
        result = json.loads(msg.data.decode())

        if "solution" in result:
            solved += 1
            print(f"[SOLVED] {result['task_id']} — {result['solution'][:30]}...")
        else:
            failed += 1
            print(f"[FAILED] {result['task_id']} — {result['error']}")

        print(f"  Stats: {solved} solved, {failed} failed")

asyncio.run(collect_results())

Dayanıklılık için NATS JetStream

Kaybolmaması gereken görevler için JetStream kalıcılığını etkinleştirin:

async def durable_publisher():
    nc = await nats.connect("nats://localhost:4222")
    js = nc.jetstream()

    # Create stream (one-time setup)
    await js.add_stream(name="CAPTCHA", subjects=["captcha.>"])

    # Publish with acknowledgment
    ack = await js.publish("captcha.tasks", json.dumps(task).encode())
    print(f"Published to stream, seq={ack.seq}")

JetStream, NATS'ın sadeliğini korurken üç yetenek ekler:

• Disk kalıcılığı — worker'lar yeniden başlasa bile görevler kaybolmaz.
• Yeniden oynatma — akışı baştan tüketip görevleri yeniden işleyebilirsiniz.
• Tam olarak bir kez teslimat — yinelenen çözüm ve gereksiz API çağrısı olmaz.

Kısacası Kafka'ya benzer dayanıklılık, ama çok daha hafif bir altyapıyla.

Worker'ları ölçeklendirme

Daha yüksek throughput için daha fazla worker başlatın; NATS kuyruk grupları yükü otomatik olarak dağıtır. Ölçeklerken iki kuralı akılda tutun:

• Thread sınırını aşmayın: worker sayısını planınızın thread sayısıyla eşleştirin — 50 thread'lik bir plan aynı anda 50 aktif çözüme izin verir, fazlası boşta bekler.
• Yatayda büyüyün: worker'ları farklı makinelere dağıtabilirsiniz; hepsi aynı kuyruk grubuna abone olduğu sürece NATS yükü paylaştırır.

# Run multiple workers — NATS distributes automatically via queue groups
python worker.py --id=1 &
python worker.py --id=2 &
python worker.py --id=3 &

# Each task goes to exactly one worker
# Add more workers to increase throughput

Sorun giderme

Sık sorulan sorular

Worker sayısını CaptchaAI plan thread'lerinize göre nasıl belirlersiniz?

Worker sayısını planınızın thread sınırına yaklaştırın: her aktif thread aynı anda bir CAPTCHA çözer. ADVANCE plan ($90/ay, 50 thread) ile 50'ye kadar eşzamanlı worker anlamlıdır; bunun ötesindeki worker'lar boşta bekler ve throughput'u artırmaz.

Bir CAPTCHA görevi çözülemezse ne olur?

Worker hatayı captcha.results konusuna bir error alanıyla yazar; toplayıcı bunu başarısız olarak sayar. Çekirdek NATS'ta bir görev kaybolursa onu yeniden yayımlarsınız — geçici iş yükü için bu davranış yeterlidir.

Aynı görevin iki worker tarafından işlenmesini nasıl önlersiniz?

Tüm worker'ları aynı kuyruk grubuna (captcha-workers) abone edin. Kuyruk grubu her mesajı yalnızca bir aboneye yönlendirir, böylece yinelenen çözüm ve gereksiz API çağrısı oluşmaz.

Üretimde JetStream'i ne zaman açmalısınız?

Yalnızca görev kaybını tolere edemediğinizde: denetim izleri, tam olarak bir kez işleme veya yeniden oynatma gerektiğinde. Çoğu CAPTCHA iş akışı için çekirdek NATS yeterlidir.

Sonraki adımlar

CAPTCHA görevlerini NATS ile dağıtın: CaptchaAI API anahtarınızı alın ve hafif worker'larınızı çalıştırın.

İlgili kılavuzlar:

• Kafka ile streaming CAPTCHA işleme
• Dağıtık işleme için Redis kuyruğu
• RabbitMQ mesaj kuyruğu entegrasyonu