RabbitMQ ile Mesaj Kuyruğu Rehberi

RabbitMQ, uygulamalar arasında mesajları kuyruklar üzerinden ileten açık kaynaklı bir mesaj aracısıdır (message broker). Üretici (producer) mesajı aracıya bırakır, tüketici (consumer) hazır olduğunda alır; iki tarafın aynı anda ayakta olması gerekmez. Bu ayrım arka plan işlerini, servisler arası olayları ve ani yük artışlarını yönetmeyi kolaylaştırır.

Kuyruk eklemek tek başına güvenilir teslim sağlamaz. Mesajın aracıya ulaştığını publisher confirms, işlendiğini tüketici onayı (acknowledgement) gösterir; işlenemeyen mesaj dead-letter akışına gider, yeniden deneme (retry) sayısı sınırlanır ve tüketici aynı mesajı iki kez almaya hazırlıklı olur. Örnekler RabbitMQ 4.x ve Python için pika 1.x istemcisiyle yazılmıştır.

RabbitMQ nasıl çalışır?

Üretici → Exchange → (binding) → Kuyruk → Tüketici

Üretici (producer): Mesajı yayımlayan uygulama. Mesajı doğrudan kuyruğa değil, bir exchange’e gönderir.
Exchange: Mesajı routing key ve binding kurallarına göre bir ya da daha çok kuyruğa yönlendiren bileşen.
Binding: Exchange ile kuyruk arasındaki yönlendirme kuralı.
Kuyruk (queue): Mesajları tüketilene kadar tutan yapı.
Tüketici (consumer): Mesajı alıp işleyen ve işlediğini aracıya onaylayan uygulama.

RabbitMQ’nun klasik protokolü AMQP 0-9-1’dir. 4.0 sürümünden beri AMQP 1.0 da eklenti gerektirmeden çekirdekte desteklenir ve iki protokol aynı aracıda birlikte çalışır. Bu yazıdaki pika istemcisi AMQP 0-9-1 kullanır.

Docker ile yerel kurulum

# docker-compose.yml: yalnız yerel geliştirme için
services:
  rabbitmq:
    image: rabbitmq:4-management
    hostname: rabbitmq # veri dizini düğüm adına, düğüm adı hostname'e bağlıdır
    ports:
      - "5672:5672" # AMQP
      - "15672:15672" # yönetim arayüzü
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: admin123 # yalnız yerel deneme şifresi
    volumes:
      - rabbitmq_data:/var/lib/rabbitmq

volumes:
  rabbitmq_data:

rabbitmq:4-management etiketi 4.x serisinin güncel sürümünü yönetim arayüzüyle getirir; üretimde 4.3-management gibi belirli bir seri sabitlenirse yükseltmeler öngörülebilir olur. Ekim 2026 itibarıyla topluluk desteği alan seri 4.3’tür, 3.x serilerinin topluluk desteği sona ermiştir. Compose v2 üst düzey version: alanını kullanmadığı için dosyada yer almaz.

Kullanıcı adı ve şifre yalnız yerel deneme içindir. Üretimde kimlik bilgilerini koda veya depoya yazmayın; ortam değişkeni ya da bir secret yöneticisiyle verin.

Python tarafında yalnız pika gerekir:

python3 -m venv .venv
source .venv/bin/activate
pip install "pika>=1.3,<2"

Güvenilir üretici ve tüketici

Aşağıdaki üç dosya birlikte çalışır: ortak bağlantı ve kuyruk tanımı, publisher confirms kullanan bir üretici ve sınırlı yeniden deneme yapan bir tüketici.

Ortak tanımlar

# rabbit.py
import os

import pika

RABBITMQ_URL = os.environ.get("RABBITMQ_URL", "amqp://admin:admin123@localhost:5672/")

ORDERS_QUEUE = "orders"
DEAD_LETTER_EXCHANGE = "orders.dlx"
DEAD_LETTER_QUEUE = "orders.dead"


def connect() -> pika.BlockingConnection:
    return pika.BlockingConnection(pika.URLParameters(RABBITMQ_URL))


def declare_topology(channel) -> None:
    # İşlenemeyen mesajların düşeceği exchange ve kuyruk
    channel.exchange_declare(
        exchange=DEAD_LETTER_EXCHANGE, exchange_type="fanout", durable=True
    )
    channel.queue_declare(queue=DEAD_LETTER_QUEUE, durable=True)
    channel.queue_bind(queue=DEAD_LETTER_QUEUE, exchange=DEAD_LETTER_EXCHANGE)

    # Ana kuyruk: replike quorum kuyruğu, teslim sınırı aşılınca DLX'e gider
    channel.queue_declare(
        queue=ORDERS_QUEUE,
        durable=True,
        arguments={
            "x-queue-type": "quorum",
            "x-dead-letter-exchange": DEAD_LETTER_EXCHANGE,
            "x-delivery-limit": 5,
        },
    )

Ana kuyruk bir quorum kuyruğudur: mesajları birden çok düğüme kopyalar ve düğümlerin çoğunluğu erişilebilir olduğu sürece hizmet verir; tek düğümlü yerel kurulumda da çalışır. x-dead-letter-exchange, reddedilen veya teslim sınırını aşan mesajın gideceği exchange’i belirler. Bu bağımsız değişken tanımlanmazsa yeniden kuyruğa alınmadan reddedilen mesaj atılır.

Kuyruk türü ve bu bağımsız değişkenler sonradan değiştirilemez. Aynı adlı kuyruğu farklı bağımsız değişkenlerle yeniden tanımlamak PRECONDITION_FAILED hatasıyla kanalı kapatır; daha önce orders adlı klasik bir kuyruk oluşturduysanız yeni bir ad seçin. RabbitMQ belgeleri, değiştirilebilir oldukları için DLX ve teslim sınırının üretimde politika (policy) ile verilmesini önerir; burada okunabilirlik için kodda tutuldu.

Publisher confirms ile üretici

# producer.py
import json
import time
import uuid

import pika
from pika.exceptions import NackError, UnroutableError

from rabbit import ORDERS_QUEUE, connect, declare_topology


def publish(channel, message_id: str, order: dict) -> None:
    channel.basic_publish(
        exchange="",  # varsayılan exchange: routing key kuyruk adıdır
        routing_key=ORDERS_QUEUE,
        body=json.dumps(order).encode("utf-8"),
        properties=pika.BasicProperties(
            content_type="application/json",
            delivery_mode=2,  # kalıcı mesaj: aracı yeniden başlasa da korunur
            message_id=message_id,
        ),
        mandatory=True,  # hiçbir kuyruğa yönlenemezse UnroutableError
    )


def publish_with_retry(channel, message_id: str, order: dict, attempts: int = 3) -> None:
    for attempt in range(1, attempts + 1):
        try:
            publish(channel, message_id, order)
            return
        except NackError:
            # Aracı mesajı kabul etmedi; aynı message_id ile tekrar dene
            if attempt == attempts:
                raise
            time.sleep(attempt)


def main() -> None:
    connection = connect()
    channel = connection.channel()
    declare_topology(channel)
    channel.confirm_delivery()  # her basic_publish aracı onaylayana kadar bekler

    orders = [
        {"order_id": "ORD001", "total": 999.99},
        {"order_id": "ORD002", "total": 29.99},
    ]
    try:
        for order in orders:
            # Kimlik olay başına bir kez üretilir, yeniden gönderimde değişmez
            message_id = str(uuid.uuid4())
            try:
                publish_with_retry(channel, message_id, order)
                print(f"Onaylandı: {order['order_id']} ({message_id})")
            except UnroutableError:
                print(f"Yönlendirilemedi, kuyruk tanımını kontrol edin: {order['order_id']}")
    finally:
        connection.close()


if __name__ == "__main__":
    main()

confirm_delivery() kanalı onay moduna alır. Bu modda pika’nın basic_publish çağrısı aracı mesajı onaylayana kadar bekler; aracı reddederse NackError, mandatory=True iken mesaj hiçbir kuyruğa yönlenemezse UnroutableError fırlatır. Kalıcı bir mesaj, dayanıklı kuyruğa diske yazıldıktan sonra; quorum kuyruğunda ise replikaların çoğunluğu kabul ettikten sonra onaylanır.

Bağlantı onaydan önce koparsa mesajın aracıya ulaşıp ulaşmadığı bilinemez. Güvenli yol, onaylanmamış mesajı yeniden bağlandıktan sonra aynı message_id ile tekrar göndermektir. Bu, aynı mesajın kuyruğa iki kez girebileceği anlamına gelir; tüketici tarafındaki idempotency kontrolü bu yüzden gereklidir.

Sınırlı yeniden deneme yapan tüketici

# consumer.py
import json

from rabbit import ORDERS_QUEUE, connect, declare_topology

# Basitleştirilmiş örnek: işlenen kimlikler bellekte tutulur.
# Gerçek sistemde benzersiz kısıtlı bir veritabanı tablosu kullanılır.
processed_ids = set()


class TemporaryError(Exception):
    """Tekrar denenince geçebilecek hata, örneğin dış servis zaman aşımı."""


def handle_order(order: dict) -> None:
    # İş mantığı burada çalışır; geçici sorunda TemporaryError fırlatılır
    print(f"İşleniyor: {order['order_id']}")


def on_message(channel, method, properties, body: bytes) -> None:
    tag = method.delivery_tag
    message_id = properties.message_id

    if message_id is None:
        # Kimliği olmayan mesajın tekrarı ayırt edilemez; incelemeye gönder
        channel.basic_reject(delivery_tag=tag, requeue=False)
        return

    if message_id in processed_ids:
        # Aynı mesaj ikinci kez geldi; yan etkiyi tekrarlamadan onayla
        channel.basic_ack(delivery_tag=tag)
        return

    try:
        handle_order(json.loads(body))
    except TemporaryError:
        # Teslim sayacını artırarak kuyruğa geri koy; sınır aşılınca DLX'e gider
        channel.basic_reject(delivery_tag=tag, requeue=True)
        return
    except (json.JSONDecodeError, KeyError):
        # Kalıcı hata: yeniden denemenin anlamı yok, doğrudan DLX'e gönder
        channel.basic_reject(delivery_tag=tag, requeue=False)
        return

    processed_ids.add(message_id)
    channel.basic_ack(delivery_tag=tag)


def main() -> None:
    connection = connect()
    channel = connection.channel()
    declare_topology(channel)
    channel.basic_qos(prefetch_count=10)  # onaylanmamış en fazla 10 mesaj
    channel.basic_consume(queue=ORDERS_QUEUE, on_message_callback=on_message)
    try:
        channel.start_consuming()
    except KeyboardInterrupt:
        channel.stop_consuming()
    finally:
        connection.close()


if __name__ == "__main__":
    main()

Tüketici auto_ack kullanmaz; mesajı ancak iş bittikten sonra onaylar. Tüketici onaydan önce çökerse mesaj kuyruğa döner ve yeniden teslim edilir; bu teslim başka bir tüketiciye de yapılabilir. RabbitMQ bu nedenle aynı mesajın birden fazla kez gelebileceğini açıkça belirtir; processed_ids kontrolü tekrar gelen mesajın yan etkisini ikinci kez üretmesini engeller. Gerçek sistemde işlenen kimliği, işin kendi yan etkisiyle aynı veritabanı işlemi (transaction) içinde ve benzersiz kısıtla kaydedin; böylece birden çok tüketici aynı anda çalışsa da kontrol tutarlı kalır.

Hata türüne göre iki farklı yol izlenir:

Geçici hata: basic_reject(requeue=True) mesajı kuyruğa geri koyar. Quorum kuyruğu başarısız teslimleri sayar ve sayaç x-delivery-limit değerini aştığında mesajı DLX’e gönderir. RabbitMQ 4.3 ve sonrasında sayacı yalnız gerçek hatalar artırır: AMQP 0-9-1’de basic.reject sayılır, basic.nack sayılmaz. Bu yüzden örnek basic_nack yerine basic_reject kullanır. Sınır verilmezse 4.0’dan beri varsayılan değer 20’dir.
Kalıcı hata: Bozuk JSON veya eksik alan tekrar denenince düzelmez. basic_reject(requeue=False) mesajı beklemeden DLX’e, oradan orders.dead kuyruğuna taşır.

requeue=True mesajı beklemeden kuyruğa döndürür. Yeniden denemeler arasında gecikme gerekiyorsa mesaj TTL’li ayrı bir bekleme kuyruğuna yönlendirilir ve süre dolunca ana kuyruğa geri döner. DLX üzerinden gelen mesajlardaki x-death başlığı mesajın hangi kuyruktan, hangi nedenle (rejected, expired, delivery_limit gibi) ve kaç kez düştüğünü gösterir.

prefetch_count, bir tüketiciye aynı anda gönderilecek onaylanmamış mesaj sayısını sınırlar. 1 en temkinli ama en yavaş ayardır; RabbitMQ belgeleri yüksek verim için 100–300 aralığından başlayıp ölçerek ayarlamayı önerir.

Exchange türleri ve yönlendirme

Exchange türü, mesajın hangi kuyruklara kopyalanacağını belirler. Aşağıdaki örnekler rabbit.py dosyasındaki connect() fonksiyonunu kullanır.

Direct exchange

Routing key ile binding key tam olarak eşleşirse mesaj o kuyruğa gider.

from rabbit import connect

connection = connect()
channel = connection.channel()
channel.exchange_declare(exchange="logs.direct", exchange_type="direct", durable=True)

for severity in ["info", "warning", "error"]:
    queue = f"logs.{severity}"
    channel.queue_declare(queue=queue, durable=True)
    channel.queue_bind(queue=queue, exchange="logs.direct", routing_key=severity)

# Yalnız "error" anahtarıyla bağlanmış logs.error kuyruğuna gider
channel.basic_publish(
    exchange="logs.direct", routing_key="error", body="Veritabanı bağlantısı koptu".encode()
)
connection.close()

Topic exchange

Topic exchange’te routing key noktayla ayrılmış kelimelerden oluşur (urgent.email.security gibi). Binding deseninde * tam olarak bir kelimeyle, # sıfır veya daha fazla kelimeyle eşleşir.

from rabbit import connect

connection = connect()
channel = connection.channel()
channel.exchange_declare(exchange="notifications", exchange_type="topic", durable=True)

bindings = {
    "notifications.email": "*.email.*",
    "notifications.sms": "*.sms.*",
    "notifications.urgent": "urgent.#",
    "notifications.user": "user.#",
}
for queue, pattern in bindings.items():
    channel.queue_declare(queue=queue, durable=True)
    channel.queue_bind(queue=queue, exchange="notifications", routing_key=pattern)

channel.basic_publish(
    exchange="notifications", routing_key="urgent.email.security", body="Güvenlik uyarısı".encode()
)
connection.close()

Bu bağlamalarla mesajlar şöyle dağılır:

Routing key	Mesajı alan kuyruklar
`user.email.welcome`	`notifications.email`, `notifications.user`
`urgent.email.security`	`notifications.email`, `notifications.urgent`
`urgent.sms.payment`	`notifications.sms`, `notifications.urgent`
`system.push.maintenance`	Hiçbiri; `mandatory=True` verilmezse mesaj atılır

Desen urgent.* olsaydı yalnız urgent.email gibi iki kelimelik anahtarları yakalardı; urgent.email.security gibi üç kelimelik anahtarlar için urgent.# gerekir.

Fanout exchange

Fanout exchange routing key’e bakmaz; mesajın bir kopyasını bağlı her kuyruğa gönderir. Her servisin kendi kuyruğu olduğu için aynı olayı her servis ayrı ayrı işler.

import json

from rabbit import connect

connection = connect()
channel = connection.channel()
channel.exchange_declare(exchange="events", exchange_type="fanout", durable=True)

for service in ["analytics", "audit", "notification"]:
    queue = f"events.{service}"
    channel.queue_declare(queue=queue, durable=True)
    channel.queue_bind(queue=queue, exchange="events")  # routing key gerekmez

event = {"type": "order_placed", "order_id": "ORD001"}
channel.basic_publish(exchange="events", routing_key="", body=json.dumps(event).encode())
connection.close()

Dördüncü tür olan headers exchange, routing key yerine mesaj başlıklarındaki değerlere göre yönlendirir; daha az kullanılır.

Üretimde dikkat edilecekler

Bağlantı ömrü: Her mesaj için bağlantı açmak pahalıdır. Süreç başına uzun ömürlü bir bağlantı açın. pika’nın BlockingConnection sınıfı iş parçacığı güvenli değildir; bağlantıyı iş parçacıkları arasında paylaşmayın.
Heartbeat: BlockingConnection ile geri çağırma (callback) içinde uzun süren bloklayıcı iş heartbeat’leri durdurur ve aracı bağlantıyı kapatabilir. Uzun işleri kısa adımlara bölün ya da ayrı bir iş parçacığına taşıyın.
Yüksek erişilebilirlik: Klasik kuyruk aynalama (mirroring) RabbitMQ 4.0 ile kaldırıldı. Replikasyon gereken kuyruklar quorum kuyruğu ya da stream olarak tanımlanmalıdır.
İzleme: Kuyruk uzunluğunu, onaylanmamış mesaj sayısını ve dead-letter kuyruğunun büyümesini izleyin. Dead-letter kuyruğunda biriken mesaj, çözülmemiş bir hatanın işaretidir.
Tüketiciyi çalıştırma: Tüketici sürekli çalışan bir süreçtir; systemd servisi olarak tanımlanırsa çöktüğünde yeniden başlatılır ve günlükleri journald’da toplanır.

RabbitMQ ile cron job farklı sorunları çözer. Cron, belirli saatte çalışacak toplu işler içindir; kuyruk ise olay geldikçe işlenmesi gereken işleri tüketicilere dağıtır. Kuyruktan gelen olayları tarayıcıya canlı iletmek gerekiyorsa tüketici bunları Server-Sent Events ile istemciye aktarabilir. Aynı desen başka sistemlerde de görülür: Mastodon gibi ActivityPub sunucuları diğer sunuculara teslimi arka plan iş kuyruklarıyla yapar.

Sonuç

RabbitMQ, iş üreten ve işleyen servisleri zaman ve hız açısından ayırmak gerektiğinde uygun bir seçimdir: arka plan işleri, servisler arası olaylar ve yük dalgalanmalarını sönümleme bu kapsama girer. Güvenilirlik kuyruğun varlığından değil; publisher confirms, manuel onay, sınırlı yeniden deneme, dead-letter akışı ve idempotent tüketiciden gelir. Mesajların uzun süre saklanıp baştan okunması gerekiyorsa RabbitMQ Streams veya log tabanlı bir sistem daha uygundur; tek makinede birkaç zamanlanmış görev için ise ayrı bir mesaj aracısı gereksiz bir katmandır.

RabbitMQ: Kuyruklar, Exchange Türleri ve Güvenilir Teslim