RabbitMQ + CaptchaAI: Mesaj Kuyruğu Entegrasyonu

Yüksek hacimli CAPTCHA çözümünü tek bir betikle yönetmeye çalıştığınızda ilk çöken şey güvenilirlik olur: worker çökerse görevler kaybolur. RabbitMQ bunu kalıcı kuyruklar, mesaj onayı (ack) ve ölü mektup (dead-letter) yönlendirmesiyle çözer. Bu rehberde CaptchaAI API'sini RabbitMQ ile üretime hazır bir producer–worker–toplayıcı hattında birleştiriyoruz.

Mimari: üç bileşen nasıl çalışır

Kurgu üç parçadan oluşur. Producer, çözülecek CAPTCHA görevlerini kalıcı captcha.tasks kuyruğuna yazar. Bir veya daha fazla worker bu kuyruğu dinler, her görevi CaptchaAI API'sine gönderir ve dönen token'ı captcha.results kuyruğuna bırakır. Sonuç toplayıcı bu kuyruğu okuyup token'ları uygulamanıza teslim eder. Başarısız görevler otomatik olarak dead-letter exchange (DLX) üzerinden captcha.failed kuyruğuna düşer; böylece hiçbir görev sessizce kaybolmaz ve sorunlu istekleri ayrı bir yerde incelersiniz. Üç bileşeni ayrı süreçlerde çalıştırmak, üretim ile tüketimi çözüm hızından bağımsız olarak ölçeklemenizi sağlar.

CAPTCHA çözümünde RabbitMQ neden işe yarar

Örnek bir senaryo: İstanbul'daki bir e-ticaret ekibi ödeme (checkout) akışını her gece test ederken kısa sürede yüzlerce reCAPTCHA doğrulamasıyla karşılaşır. RabbitMQ ile görevler kuyrukta bekler, boşta kalan worker sıradakini alır. CaptchaAI'nin thread tabanlı planları bu modele oturur: her eşzamanlı çözüm bir thread tüketir, yani worker sayınızı thread sayınızla hizalarsınız. Kişisel veri içeren akışlarda çalışmayı KVKK kapsamında yetkili QA ortamlarıyla sınırlayın.

Redis yerine ne zaman RabbitMQ tercih edilir?

İki seçenek de kuyruk sağlar ama farklı önceliklere hitap eder. Redis, düşük gecikme ve çok basit bir kurulum isteyen, tek bir görevin kaybolmasının kritik olmadığı işler için idealdir. RabbitMQ ise mesaj onayı, dead-letter yönlendirmesi ve yönlendirme anahtarlarıyla teslim güvencesi sunar. Ödeme QA'sı gibi her görevin işlenmesi gereken ve farklı CAPTCHA türlerini ayrı worker havuzlarına dağıtmak istediğiniz senaryolarda RabbitMQ daha uygundur. Küçük ve tek türlü bir iş yükü içinse Redis'in sadeliği çoğu zaman yeterlidir.

Kurulum

RabbitMQ ve Python istemcisini kurun:

# Docker
docker run -d --hostname rabbitmq \
  -p 5672:5672 -p 15672:15672 \
  rabbitmq:3-management

# Python client
pip install pika requests

Yönetim arayüzü 15672 portunda açılır; kuyruk uzunluklarını buradan izlersiniz.

Producer: görevleri kuyruğa gönderin

Producer, kuyrukları ve DLX'i bir kez tanımlar, ardından her görevi delivery_mode=2 ile diske yazarak broker yeniden başlasa bile kaybolmamasını sağlar. priority alanı, kullanıcıyı bekleten acil doğrulamaları toplu işlerin önüne almanıza olanak tanır; x-message-ttl ise bir görevin kuyrukta ne kadar bekleyeceğini sınırlar.

import json
import uuid
import pika


class CaptchaProducer:
    """Submit CAPTCHA tasks to RabbitMQ."""

    def __init__(self, rabbitmq_url="amqp://guest:guest@localhost:5672/"):
        self.connection = pika.BlockingConnection(
            pika.URLParameters(rabbitmq_url),
        )
        self.channel = self.connection.channel()
        self._setup_queues()

    def _setup_queues(self):
        """Declare durable queues and exchanges."""
        # Dead letter exchange for failed tasks
        self.channel.exchange_declare(
            exchange="captcha.dlx",
            exchange_type="direct",
            durable=True,
        )
        self.channel.queue_declare(
            queue="captcha.failed",
            durable=True,
        )
        self.channel.queue_bind(
            queue="captcha.failed",
            exchange="captcha.dlx",
            routing_key="failed",
        )

        # Main task queue with dead letter routing
        self.channel.queue_declare(
            queue="captcha.tasks",
            durable=True,
            arguments={
                "x-dead-letter-exchange": "captcha.dlx",
                "x-dead-letter-routing-key": "failed",
                "x-message-ttl": 300000,  # 5 min TTL
            },
        )

        # Results queue
        self.channel.queue_declare(
            queue="captcha.results",
            durable=True,
        )

    def submit(self, method, params, priority=0):
        """Submit a CAPTCHA task."""
        task_id = str(uuid.uuid4())[:8]
        task = {
            "id": task_id,
            "method": method,
            "params": params,
        }

        self.channel.basic_publish(
            exchange="",
            routing_key="captcha.tasks",
            body=json.dumps(task),
            properties=pika.BasicProperties(
                delivery_mode=2,  # Persistent
                priority=priority,
                message_id=task_id,
            ),
        )
        return task_id

    def close(self):
        self.connection.close()


# Usage
producer = CaptchaProducer()

task_id = producer.submit("userrecaptcha", {
    "googlekey": "SITE_KEY",
    "pageurl": "https://example.com",
}, priority=5)

print(f"Submitted: {task_id}")
producer.close()

Consumer: worker CAPTCHA'ları çözer

Worker aynı anda tek görev işler (prefetch_count=1), görevi CaptchaAI API'sine in.php üzerinden gönderir ve token'ı res.php sorgulamasıyla alır. Başarılı sonuç captcha.results kuyruğuna yazılır ve mesaj basic_ack ile onaylanır; bir hata çıkarsa basic_nack çağrısı görevi DLX'e yönlendirir. prefetch_count=1, uzun süren bir çözümün sıradaki görevleri bloke etmesini engeller.

import json
import os
import time
import pika
import requests


class CaptchaConsumer:
    """RabbitMQ consumer that solves CAPTCHAs."""

    def __init__(self, api_key, rabbitmq_url="amqp://guest:guest@localhost:5672/"):
        self.api_key = api_key
        self.base = "https://ocr.captchaai.com"
        self.connection = pika.BlockingConnection(
            pika.URLParameters(rabbitmq_url),
        )
        self.channel = self.connection.channel()
        # Process one task at a time
        self.channel.basic_qos(prefetch_count=1)

    def start(self):
        """Start consuming tasks."""
        self.channel.basic_consume(
            queue="captcha.tasks",
            on_message_callback=self._handle_task,
        )
        print("Worker started. Waiting for tasks...")
        self.channel.start_consuming()

    def _handle_task(self, ch, method, properties, body):
        """Process a single CAPTCHA task."""
        task = json.loads(body)
        task_id = task["id"]
        print(f"Processing {task_id}...")

        try:
            token = self._solve(task["method"], task["params"])

            # Publish result
            result = {
                "task_id": task_id,
                "status": "success",
                "token": token,
            }
            ch.basic_publish(
                exchange="",
                routing_key="captcha.results",
                body=json.dumps(result),
                properties=pika.BasicProperties(delivery_mode=2),
            )

            # Acknowledge message (remove from queue)
            ch.basic_ack(delivery_tag=method.delivery_tag)
            print(f"{task_id} solved successfully")

        except Exception as e:
            print(f"{task_id} failed: {e}")

            # Reject and send to dead letter queue
            ch.basic_nack(
                delivery_tag=method.delivery_tag,
                requeue=False,  # Goes to DLX
            )

    def _solve(self, captcha_method, params, timeout=120):
        resp = requests.post(f"{self.base}/in.php", data={
            "key": self.api_key,
            "method": captcha_method,
            "json": 1,
            **params,
        }, timeout=30)
        result = resp.json()

        if result.get("status") != 1:
            raise RuntimeError(result.get("request"))

        captcha_id = result["request"]
        start = time.time()

        while time.time() - start < timeout:
            time.sleep(5)
            resp = requests.get(f"{self.base}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": captcha_id,
                "json": 1,
            }, timeout=15)
            data = resp.json()
            if data["request"] != "CAPCHA_NOT_READY":
                if data.get("status") == 1:
                    return data["request"]
                raise RuntimeError(data["request"])

        raise TimeoutError("Solve timeout")


# Run worker
if __name__ == "__main__":
    consumer = CaptchaConsumer(os.environ["CAPTCHAAI_KEY"])
    consumer.start()

Sonuçları toplayın

Toplayıcı, captcha.results kuyruğundan beklenen sayıda yanıtı toplar. Onu ayrı bir süreçte çalıştırmak çözüm mantığı ile sonuç işlemeyi birbirinden ayırır ve toplayıcıyı yeniden başlatmadan worker'ları ölçeklemenize izin verir.

import json
import pika


class ResultCollector:
    """Collect task results from the results queue."""

    def __init__(self, rabbitmq_url="amqp://guest:guest@localhost:5672/"):
        self.connection = pika.BlockingConnection(
            pika.URLParameters(rabbitmq_url),
        )
        self.channel = self.connection.channel()
        self.results = {}

    def collect(self, expected_count, timeout=120):
        """Collect a specific number of results."""
        deadline = time.time() + timeout

        while len(self.results) < expected_count and time.time() < deadline:
            method, _, body = self.channel.basic_get(
                queue="captcha.results",
                auto_ack=True,
            )
            if body:
                result = json.loads(body)
                self.results[result["task_id"]] = result

            time.sleep(0.5)

        return self.results

CAPTCHA türüne göre yönlendirme

CAPTCHA türünü yönlendirme anahtarı yaparak reCAPTCHA, Turnstile ve görüntü tabanlı görevleri ayrı kuyruklara dağıtabilirsiniz. Böylece her tür kendi worker havuzunda bağımsız ölçeklenir; örneğin görüntü/OCR görevleri için ayrı, reCAPTCHA için ayrı worker grupları çalıştırırsınız.

# Setup exchanges and queues
channel.exchange_declare(
    exchange="captcha.types",
    exchange_type="direct",
    durable=True,
)

# Queue per type
for captcha_type in ["recaptcha", "turnstile", "image"]:
    channel.queue_declare(queue=f"captcha.{captcha_type}", durable=True)
    channel.queue_bind(
        queue=f"captcha.{captcha_type}",
        exchange="captcha.types",
        routing_key=captcha_type,
    )


# Submit with routing
def submit_routed(channel, captcha_type, task):
    channel.basic_publish(
        exchange="captcha.types",
        routing_key=captcha_type,
        body=json.dumps(task),
        properties=pika.BasicProperties(delivery_mode=2),
    )

Yeniden deneme ve başarısız görevler

Geçici hatalar (zaman aşımı, kısa ağ kesintisi) ile kalıcı hatalar farklı ele alınmalıdır. Başarısız bir görevi hemen atmak yerine, TTL gecikmeli bir yeniden deneme (retry) exchange'i kurun: worker tarafından reddedilen mesaj kısa bir süre bekletilir ve otomatik olarak captcha.tasks kuyruğuna geri alınır. Belirli bir deneme sayısından sonra hâlâ çözülemeyen görevler captcha.failed kuyruğunda birikir. Bu kuyruğu düzenli inceleyin; hızla büyümesi çoğunlukla yanlış sitekey/pageurl parametrelerine ya da süresi dolmuş görevlere işaret eder ve genelde bir kod değil yapılandırma sorunudur.

İzleme ve ölçekleme

Üretimde iki metriği takip edin: captcha.tasks kuyruk uzunluğu ve captcha.failed büyüme hızı. Yönetim arayüzü (15672 portu) her ikisini de gösterir. Kuyruk sürekli birikiyorsa worker kapasiteniz yetersizdir. CaptchaAI'de eşzamanlılık thread sayısıyla belirlendiği için, worker sayınızı planınızın thread sayısıyla eşleştirin:

Daha fazla eşzamanlı worker'a ihtiyaç duydukça bir üst plana geçmeniz yeterli; kuyruk mimarisinde ya da kodda değişiklik gerekmez. USD üzerinden sabit aylık ücret, TL kur oynaklığına karşı öngörülebilir bir maliyet sağlar.

Sorun giderme

Üretim kontrol listesi

Hattı canlıya almadan önce şu maddeleri doğrulayın:

• Tüm kuyrukları durable=True, tüm mesajları delivery_mode=2 ile tanımlayın.
• Her worker için prefetch_count=1 ayarlayın.
• DLX ve captcha.failed kuyruğunu izlemeye alın.
• Worker sayısını planınızın thread sayısıyla sınırlayın.
• Sonuç işleyicinizi task_id üzerinden idempotent yapın.
• Bağlantılar için heartbeat ve yeniden bağlanma mantığı ekleyin.
• Kişisel veri içeren akışları yalnızca KVKK kapsamında yetkili QA ortamlarında test edin.

Sık sorulan sorular

CaptchaAI thread sayısı ile worker sayısını nasıl eşleştiririm?

Her worker aynı anda tek CAPTCHA çözer ve bir thread tüketir; bu yüzden worker sayısını plan thread sayısıyla sınırlayın. STANDARD ($30/ay, 15 thread) 15 worker besler.

Acil doğrulamaları önce çözmek için öncelik kuyruklarını nasıl kullanırım?

Kuyruğu x-max-priority ile tanımlayın ve submit çağrısında yüksek bir priority verin. Böylece kullanıcıyı bekleten canlı doğrulama, toplu işlerin önüne geçer.

Dead-letter (DLX) kuyruğunu nasıl izlerim?

Yönetim arayüzünde (15672) captcha.failed kuyruğunun uzunluğunu izleyin. Sürekli büyüyorsa kalıcı bir hata vardır; genelde yanlış sitekey/pageurl ya da zaman aşımı.

Aynı CAPTCHA iki kez çözülür mü?

Worker mesajı sonucu yazdıktan sonra basic_ack ile onaylar. Onaydan önce çökme olursa RabbitMQ mesajı yeniden teslim edebilir; sonuç işleyicinizi task_id üzerinden idempotent tutun.

İlgili rehberler

• Redis kuyruğu ile dağıtık CAPTCHA işleme
• Birden fazla görevi toplu CAPTCHA çözme

Kuyrukları RabbitMQ, çözümü CaptchaAI yönetsin — ücretsiz hesabınızı açın.