RabbitMQ Topology'si Kodda: Exchange Tipleri, Binding, Quorum Queue ve Stream

Sürüm notu: Bu yazıdaki kodlar .NET 10, RabbitMQ 4.4 ve RabbitMQ.Client 7.2.2 ile derlenip denenmiştir.

Önce Tahmin Et

bookstore.ordering adında bir topic exchange'imiz ve ona bağlı dört queue'muz var. Binding key'ler şöyle: billing.order-placed için order.placed, notification.order-placed için yine order.placed, analytics.order-status için order.*, audit.orders için order.#. Exchange'e dört mesaj gönderiyoruz; routing key'leri order.placed, order.cancelled, order.shipped ve order.shipped.partial. Her queue'da kaç mesaj olur? Toplamda kaç kopya oluşur?

Bir de şu: billing.order-placed zaten quorum queue olarak var. Başka bir servis aynı adı argümansız, yani classic queue olarak declare ederse ne olur? Hata mı alır, yoksa queue'nun tipi mi değişir?

Exchange, Binding, Queue: Mesajın Yolu

RabbitMQ'da producer queue'ya yazmaz, exchange'e yazar. Exchange mesajın routing key'ine ve kendi tipine bakar, binding tablosundaki eşleşen queue'lara mesajın bir kopyasını koyar. Hiçbir binding eşleşmezse mesaj düşer; önceki yazıda bunu mandatory bayrağı ve PublishReturnException ile yakalamıştık. Queue, point-to-point ve publish/subscribe kavramlarının mimari tarafı Messaging ve Event-Driven Architecture serisinin "Channel'lar: Point-to-Point Queue, Publish/Subscribe Topic ve Competing Consumer'lar" yazısında. Burada RabbitMQ'da bunun nasıl kurulduğuna bakıyoruz.

Önce gözden kaçan bir exchange: adı boş string olan default exchange. Her queue ona, kendi adıyla aynı routing key ile otomatik bağlıdır. Önceki yazıdaki deneylerde BasicPublishAsync("", "lab.perf", body) yazdığımızda mesajı bu exchange üzerinden doğrudan lab.perf queue'suna gönderiyorduk. Hızlı denemeler için pratik, ama producer'ı consumer'ın queue adına bağladığı için gerçek akışlarda kendi exchange'ini kullan.

Topic Exchange ve Routing Key Tasarımı

Bookstore'un sipariş event'leri için topology'yi tek bir class'ta topluyoruz:

namespace Bookstore.Messaging;

public static class BookstoreTopology
{
    public const string Ordering = "bookstore.ordering";

    private static readonly Dictionary<string, object?> Quorum = new() { ["x-queue-type"] = "quorum" };

    public static async Task DeclareAsync(IChannel channel, CancellationToken ct = default)
    {
        await channel.ExchangeDeclareAsync(Ordering, ExchangeType.Topic, durable: true, cancellationToken: ct);

        await BindQueueAsync(channel, "billing.order-placed", "order.placed", ct);
        await BindQueueAsync(channel, "notification.order-placed", "order.placed", ct);
        await BindQueueAsync(channel, "analytics.order-status", "order.*", ct);
        await BindQueueAsync(channel, "audit.orders", "order.#", ct);
    }

    private static async Task BindQueueAsync(IChannel channel, string queue, string bindingKey, CancellationToken ct)
    {
        await channel.QueueDeclareAsync(queue, durable: true, exclusive: false, autoDelete: false,
            arguments: Quorum, cancellationToken: ct);
        await channel.QueueBindAsync(queue, Ordering, bindingKey, cancellationToken: ct);
    }
}

Dört mesajı gönderdikten sonra rabbitmqctl list_queues name type messages çıktısı:

audit.orders	quorum	4
notification.order-placed	quorum	1
billing.order-placed	quorum	1
analytics.order-status	quorum	3

Dört mesajdan dokuz kopya çıktı. Topic exchange'te routing key noktalarla ayrılmış kelimelerden oluşur ve binding key'de iki joker var: * tam olarak bir kelimeyle, # sıfır veya daha fazla kelimeyle eşleşir. order.shipped.partial üç kelime olduğu için order.*'a uymuyor, order.#'a uyuyor. Billing ve Notification aynı binding key'e sahip oldukları için ikisi de kendi kopyasını alıyor; publish/subscribe'ın RabbitMQ'daki hâli bu: her abone kendi queue'su.

Routing key'i bir isim alanı gibi tasarla. Bookstore'da kural <entity>.<event>[.<qualifier>], yani entity, event ve isteğe bağlı bir niteleyici: order.placed, order.cancelled, order.shipped.partial. Birkaç pratik kural:

• Genelden özele sırala. order.placed.tr sorgulanabilir (order.placed.*, order.#), tr.placed.order sorgulanamaz hâle gelir.
• Routing key'e değişken veri koyma. Sipariş id'si veya müşteri e-postası routing key'de olursa binding'ler anlamsızlaşır; bu bilgiler mesaj gövdesinde ya da header'da durur.
• Sürümü routing key'e değil mesajın type property'sine koy. Contract sürümlemeyi serinin contract yazısında konuşacağız.
• Consumer'lar dar binding ile başlasın. # ile bağlanan bir queue, ileride eklenen her yeni event'i de alır; audit gibi gerçekten her şeyi isteyen tüketiciler dışında bu bir sürprizdir.

Direct, Fanout ve Headers Exchange

Direct exchange routing key'in binding key ile birebir aynı olmasını ister; joker yok. Komut tarzı mesajlar (tek bir hedefe giden "fatura kes" gibi) için temiz bir seçim. Topic exchange joker kullanılmadığında direct gibi davrandığı için birçok ekip event'lerde baştan topic kullanır; ileride joker gerekirse exchange'i değiştirmek zorunda kalmaz.

Fanout exchange routing key'e hiç bakmaz, bağlı her queue'ya kopya bırakır. Bookstore'da bunu cache invalidation için kullanıyoruz: bir kitabın fiyatı değiştiğinde API'nin (Application Programming Interface) her instance'ı kendi bellek içi cache'ini temizlemeli. Her instance kendine geçici bir queue açar:

const string exchange = "bookstore.catalog.changed";
for (int instance = 1; instance <= 2; instance++)
{
    IChannel ch = await connection.CreateChannelAsync();
    await ch.ExchangeDeclareAsync(exchange, ExchangeType.Fanout, durable: true);
    // her instance kendi geçici queue'sunu açar: sunucu adlandırır, bağlantı kapanınca silinir
    QueueDeclareOk q = await ch.QueueDeclareAsync(queue: "", durable: false, exclusive: true, autoDelete: true);
    await ch.QueueBindAsync(q.QueueName, exchange, routingKey: "");
    int id = instance;
    AsyncEventingBasicConsumer consumer = new(ch);
    consumer.ReceivedAsync += (_, ea) =>
    {
        Console.WriteLine($"instance-{id} ({q.QueueName}) got: {Encoding.UTF8.GetString(ea.Body.Span)}");
        return Task.CompletedTask;
    };
    await ch.BasicConsumeAsync(q.QueueName, autoAck: true, consumer);
}
await using IChannel pub = await connection.CreateChannelAsync();
await pub.BasicPublishAsync(exchange, "", Encoding.UTF8.GetBytes("book 42 price changed"));

instance-1 (amq.gen-XmdDueBrTfu8FFCsl84_fw) got: book 42 price changed
instance-2 (amq.gen-i58KEXRtTEK1gWU7Qsg22Q) got: book 42 price changed

Boş isimle declare edilen queue'ya sunucu amq.gen-... adı veriyor. exclusive: true queue'yu yalnızca bu bağlantıya ait kılıyor ve bağlantı kapanınca siliyor. Burada autoAck: true bilinçli: cache temizleme mesajı kaybolursa en kötü ihtimalle cache TTL'i (Time To Live) dolunca düzelir, ve instance zaten ölürse temizlenecek cache'i de kalmaz. Bu queue'lar quorum olamaz; quorum queue exclusive ve geçici olmayı desteklemiyor, bu iş classic queue'nun.

Headers exchange routing key yerine mesaj header'larına bakar. Binding'de x-match ile "hepsi eşleşsin" (all) ya da "biri eşleşsin" (any) seçilir:

await ch.QueueBindAsync("shipping.express-eu", exchange, routingKey: "",
    arguments: new Dictionary<string, object?> { ["x-match"] = "all", ["region"] = "eu", ["speed"] = "express" });

foreach ((string region, string speed) in new[] { ("eu", "express"), ("eu", "standard"), ("tr", "express") })
{
    BasicProperties props = new() { Headers = new Dictionary<string, object?> { ["region"] = region, ["speed"] = speed } };
    await ch.BasicPublishAsync(exchange, "", mandatory: false, props, Encoding.UTF8.GetBytes($"{region}/{speed}"));
}

shipping.express-eu	quorum	1

Üç mesajdan yalnızca eu/express geldi. Headers exchange esnek görünür ama pratikte nadiren gerekir: aynı şeyi shipping.express.eu gibi bir routing key ve topic exchange ile yapmak hem daha hızlı hem de management arayüzünde daha okunur.

Topology'yi Idempotent Declare Etmek

Her servis kendi kullandığı exchange ve queue'ları startup'ta declare eder. Producer exchange'i, consumer exchange'i, kendi queue'sunu ve binding'ini. Böylece servislerin başlama sırası önemsizleşir. Declare idempotent'tir, ama tek şartla: aynı isim aynı argümanlarla gelmeli. Yazının başındaki ikinci sorunun cevabı:

// aynı isim, farklı argüman: classic queue olarak declare etmeye çalış
await ch.QueueDeclareAsync("billing.order-placed", durable: true, exclusive: false, autoDelete: false);

The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=406, text='PRECONDITION_FAILED - inequivalent arg 'x-queue-type' for queue 'billing.order-placed' in vhost '/': received 'classic' but current is 'quorum'', classId=50, methodId=10
Channel open after error: False

Queue'nun tipi değişmedi; declare eden channel kapandı. Bu iyi bir şey: iki servis aynı queue hakkında farklı fikirdeyse bunu startup'ta, yüksek sesle öğrenirsin. Ama iki sonucu var. Birincisi, channel-level hatalar channel'ı öldürür; topology declare'ı ayrı, kısa ömürlü bir channel'da yap ve publish channel'ını riske atma. İkincisi, queue argümanları değiştirilemez. Bir queue'nun tipini ya da x-max-length argümanını değiştirmek için queue'yu silip yeniden oluşturman gerekir, yani içindeki mesajları başka yere taşıman. Bu yüzden değişebilecek ayarlar argümanda değil policy'de durmalı; birazdan oraya geleceğiz.

Queue Tipleri: Classic, Quorum ve Stream

RabbitMQ 4.x'te üç queue tipi var ve farkları sadece performans değil, davranış:

Kalıcı iş mesajları için varsayılan seçim quorum queue. Tek node'lu geliştirme ortamında replikasyon yapmaz ama kod ve davranış production cluster'ıyla aynı kalır. delivery-limit özellikle değerli: bir mesaj 20 kez teslim edilip her seferinde nack'lenirse quorum queue onu düşürür ya da dead-letter ayarı varsa DLX'e (Dead Letter Exchange) gönderir. Classic queue'da aynı mesaj sonsuza kadar döner. Bu limitin retry stratejisiyle ilişkisini serinin retry yazısında kuracağız.

Stream bir log'dur: mesajlar ack'lendikten sonra da durur, retention (x-max-age, x-max-length-bytes) dolunca segment segment silinir. Her consumer kendi offset'inden okur. Audit için bir stream:

await ch.QueueDeclareAsync("audit.order-stream", durable: true, exclusive: false, autoDelete: false,
    arguments: new Dictionary<string, object?> { ["x-queue-type"] = "stream", ["x-max-age"] = "7D" });

for (int reader = 1; reader <= 2; reader++)
{
    await using IChannel rc = await connection.CreateChannelAsync();
    // stream consumer'ı için prefetch zorunlu
    await rc.BasicQosAsync(0, 100, false);
    List<string> seen = [];
    AsyncEventingBasicConsumer consumer = new(rc);
    consumer.ReceivedAsync += async (_, ea) =>
    {
        seen.Add(Encoding.UTF8.GetString(ea.Body.Span));
        await rc.BasicAckAsync(ea.DeliveryTag, false);
    };
    await rc.BasicConsumeAsync("audit.order-stream", autoAck: false, consumerTag: "", noLocal: false,
        exclusive: false, arguments: new Dictionary<string, object?> { ["x-stream-offset"] = "first" },
        consumer: consumer);
    await Task.Delay(700);
    Console.WriteLine($"reader-{reader} read {seen.Count}: {string.Join(", ", seen)}");
}

Published 5 messages to the stream
reader-1 read 5: order-1, order-2, order-3, order-4, order-5
reader-2 read 5: order-1, order-2, order-3, order-4, order-5

İkinci okuyucu, birincinin ack'lediği mesajları baştan okudu. x-stream-offset için first, last, next, sayısal bir offset ya da bir zaman damgası verilebilir. AMQP (Advanced Message Queuing Protocol) 0-9-1 üzerinden stream okumak için prefetch ve manual ack zorunlu; ack burada mesajı silmez, sadece akış kontrolü için kredi verir. Yüksek throughput gerekiyorsa RabbitMQ dokümantasyonu stream'e özel binary protokolü ve onun client'larını (.NET için RabbitMQ.Stream.Client) öneriyor. Replay ihtiyacı ağır basıyorsa Kafka da masada; karşılaştırmayı serinin ilk yazısında yapmıştık.

TTL, Max-Length ve Overflow

Queue'lar sonsuz büyüyemez. Consumer saatlerce düşük kaldığında queue'nun broker'ın belleğini ve diskini doldurmasını istemezsin. İki ana araç var: mesaja ömür vermek (x-message-ttl, ya da mesaj bazında Expiration property'si) ve queue'ya boy sınırı koymak (x-max-length, x-max-length-bytes). Sınır dolduğunda ne olacağını x-overflow belirler: varsayılan drop-head en eski mesajı atar, reject-publish yeni mesajı reddeder. Sipariş gibi kaybı kabul edilemeyen mesajlarda eskiyi sessizce atmak yerine producer'a "dur" demek daha dürüst:

await ch.QueueDeclareAsync("lab.limited", durable: true, exclusive: false, autoDelete: false,
    arguments: new Dictionary<string, object?>
    {
        ["x-queue-type"] = "quorum",
        ["x-max-length"] = 3,
        ["x-overflow"] = "reject-publish"
    });

m1: confirmed
m2: confirmed
m3: confirmed
m4: confirmed
m5: PublishException IsReturn=False
m6: PublishException IsReturn=False

İki ayrıntı var. Birincisi, reddi yalnızca publisher confirmation açıksa görürsün; broker bir basic.nack gönderir ve client bunu IsReturn=False olan bir PublishException'a çevirir. Confirm kapalı bir producer'da mesaj sessizce kaybolur. İkincisi, bizim denememizde quorum queue limiti tam sınırda değil, bir mesaj geç uyguladı: dördüncü mesaj kabul edildi, beşinci reddedildi. Limitleri kesin bir sayı değil, bir emniyet kemeri gibi düşün.

Client mı, Policy mi?

Bir ayarı kodda queue argümanı olarak mı vermeli, yoksa broker'da policy olarak mı tanımlamalı? Policy, isim kalıbına (regex) uyan queue'lara uygulanan ve çalışma anında değiştirilebilen bir ayar setidir:

docker exec mqwork1-rabbit rabbitmqctl set_policy --apply-to quorum_queues --priority 10 billing-limits '^billing\.' '{"delivery-limit":5,"max-length":100000,"overflow":"reject-publish"}'
docker exec mqwork1-rabbit rabbitmqctl list_queues name policy effective_policy_definition arguments

name	policy	effective_policy_definition	arguments
audit.orders		#{}	[{"x-queue-type","quorum"}]
billing.order-placed	billing-limits	[{<<"delivery-limit">>, 5}, {<<"max-length">>, 100000}, {<<"overflow">>, reject-publish}]	[{"x-queue-type","quorum"}]

Queue yeniden declare edilmeden yeni limitleri aldı. Pratik ayrım şöyle:

• Kodda (argüman): queue'nun kimliği ve uygulamanın doğruluğu için şart olanlar. Queue tipi (x-queue-type), durable, exclusive, stream'in varlığı.
• Policy'de: ortama göre değişen ve çalışırken ayarlanması gerekebilecekler. TTL, max-length, overflow, delivery-limit, dead-letter exchange.

Öncelik kurallarını bil. RabbitMQ dokümantasyonuna göre aynı anahtar hem argümanda hem kullanıcı policy'sinde varsa genel kural olarak argüman kazanır; max-length için ise ikisinin küçüğü uygulanır. Bir queue'ya aynı anda yalnızca bir (kullanıcı) policy'si uygulanır; kalıpları çakışan iki policy'den önceliği yüksek olan kazanır. Operatörlerin ezilemez sınırlar koyması için ayrıca operator policy'leri var. Yani kodda x-max-length verip sonra policy ile büyütmeye çalışmak işe yaramaz. Değişecek ayarı baştan policy'ye bırak.

Üretimde Dikkat

Topology'yi kim declare ediyor sorusunun bir cevabı olsun. Küçük sistemlerde "her servis kendi ihtiyacını declare eder" yeterli. Büyük sistemlerde topology'yi bir definitions dosyası (rabbitmqctl export_definitions) ya da altyapı kodu ile yönetip uygulamaların declare yetkisini kısıtlamak, yanlış argümanla açılan queue'ları önler.

Queue sayısını izle. Her quorum queue ayrı bir Raft grubu; binlerce quorum queue cluster'a ciddi yük bindirir. Müşteri veya sipariş başına queue açmak yerine routing key ve binding ile ayrıştır.

# ile bağlı queue'ları izle. audit.orders gibi her şeyi alan bir queue, consumer'ı durduğunda en hızlı büyüyen queue olur. Ona mutlaka max-length ya da stream gibi bir retention sınırı koy.

Dead-letter'ı topology ile birlikte tasarla. Quorum queue'nun delivery-limit'i dolduğunda bir DLX tanımlı değilse mesaj sessizce silinir. DLX ve DLQ'nun (Dead Letter Queue) nasıl yönetileceğini serinin dead-letter yazısında ayrıntılı kuracağız; şimdilik policy'de bir dead-letter-exchange bulunsun.

Sık Düşülen Tuzaklar

Classic queue'yu "varsayılan" diye kullanmak. QueueDeclareAsync'e argüman vermezsen classic queue alırsın; 4.x'te bu replikasyonsuz bir queue demek. Kalıcı iş mesajları için x-queue-type'ı açıkça yaz.

Queue argümanını sonradan değiştirmeye çalışmak. PRECONDITION_FAILED ile karşılaşırsın ve channel kapanır. Değişebilecek her şeyi policy'ye taşı.

Routing key'e değişken veri koymak. order.placed.01a1b8de... gibi key'ler binding'leri anlamsızlaştırır ve management arayüzünde okunamaz hâle gelir.

Fanout ile her instance'a mesaj dağıtırken quorum queue denemek. Exclusive, geçici queue'lar classic olmak zorunda.

Stream'i queue gibi tüketmek. Ack mesajı silmez; offset'i kaydetmezsen consumer her yeniden başladığında first ile her şeyi baştan işler ya da next ile aradaki mesajları kaçırır.

reject-publish'i confirm'süz kullanmak. Broker reddeder, producer haberdar olmaz, mesaj kaybolur.

Küçük Bir Deneme

★☆☆ order.placed.tr routing key'i ile bir mesaj gönder. Yukarıdaki dört queue'dan hangileri alır? Önce tahmin et, sonra rabbitmqctl list_queues ile kontrol et.

★★☆ billing.order-placed için message-ttl ve dead-letter-exchange içeren bir policy tanımla. Consumer kapalıyken mesaj gönder ve TTL dolunca mesajın nereye gittiğini izle.

★★★ audit.order-stream'i okuyan bir consumer yaz ve son işlediği offset'i (mesajın x-stream-offset header'ından) bir dosyaya kaydetsin. Process yeniden başladığında kaldığı offset'ten devam etsin ve hiçbir mesajı iki kez işlemediğini göster.

RabbitMQ'da topology'yi kodla ve policy ile kurmayı gördük. Bir sonraki yazıda yönetilen bir broker'a geçiyoruz: Azure Service Bus'ın .NET SDK'sı (Software Development Kit) ile queue, topic ve subscription, ServiceBusProcessor, PeekLock ve lock süresi, session'larla sıralı işlem ve hepsini yerelde Service Bus emulator'ü ile denemek.