Channel'lar: Point-to-Point Queue, Publish/Subscribe Topic ve Competing Consumer'lar

Sürüm notu: Bu yazıdaki kodlar C# 14 ve .NET 10 ile, RabbitMQ 4.4 ve Apache Kafka 4.2 (Docker container'ları) üzerinde derlenip denenmiştir.

Önce Tahmin Et

Bookstore OrderPlaced'i artık bir broker'a yayınlıyor. Inventory'nin bir consumer'ı var. Shipping, kampanya yükü yüzünden consumer'ını iki instance'a çıkardı. Altı sipariş geliyor.

Üç soru. Inventory kaç message görür, Shipping'in iki instance'ı kaç message'ı paylaşır? Bir hafta sonra Analytics ekibi aynı event'leri dinlemek için yeni bir queue açarsa, o queue'da geçen haftanın siparişlerinden kaç tane bulur? Aynı soruyu Kafka için sorarsak cevap değişir mi?

Channel: Message'ın Yolu

Bir önceki yazıda message'ın kendisine baktık. Enterprise Integration Patterns kitabı (EIP: Gregor Hohpe ve Bobby Woolf, 2003) message'ın taşındığı yola message channel diyor: producer'ı consumer'a bağlayan, mantıksal bir adres. Kitap channel'ları davranışına göre ikiye ayırıyor. Point-to-Point Channel'da her message'ı tek bir consumer alır. Publish-Subscribe Channel'da ise her subscriber message'ın kendi kopyasını alır. Bu ayrım bir önceki yazının command/event ayrımıyla doğrudan eşleşiyor: command, tek alıcıya giden point-to-point bir channel'a gönderilir; event, kimin dinlediğini bilmeyen bir publish/subscribe channel'ına yayınlanır.

Kitap, aynı channel'dan okuyan birden fazla consumer'a da bir ad veriyor: Competing Consumers. Consumer'lar message'lar için yarışır, her message yalnızca birine gider. Bir queue'nun ölçeklenmesinin temel yolu bu: iş yetişmiyorsa consumer instance'ı eklersin.

Kavramlar broker'dan bağımsız, ama her broker onları farklı nesnelerle kuruyor. Aynı OrderPlaced'i iki broker'a, RabbitMQ'ya ve Kafka'ya yayınlayıp farkı çalışan kodda görelim. İki adapter'ın da arkasında duracağı soyutlama küçük:

namespace Bookstore.Ordering.Contracts;

public interface IIntegrationEvent
{
    static abstract string MessageType { get; }
}

public sealed record OrderPlaced(Guid OrderId, string Customer, string Title, int Quantity, decimal Total)
    : IIntegrationEvent
{
    public static string MessageType => "bookstore.ordering.order-placed.v1";
}

using Bookstore.Ordering.Contracts;

namespace Bookstore.Messaging;

// Uygulama yalnızca bunu görür: "bu event'i yayınla"
public interface IMessagePublisher
{
    Task PublishAsync<T>(T message, string key, CancellationToken ct = default) where T : IIntegrationEvent;
}

using Bookstore.Messaging;
using Bookstore.Ordering.Contracts;

namespace Bookstore.Ordering;

// Hangi broker'ın kullanıldığını bilmiyor
public sealed class PlaceOrderHandler(IMessagePublisher publisher)
{
    public Task HandleAsync(OrderPlaced placed, CancellationToken ct = default) =>
        publisher.PublishAsync(placed, key: placed.OrderId.ToString(), ct);
}

Bu yazıdaki client kodu kavramı göstermek için yeterli, ama production için değil: connection yönetimi, publisher confirm, retry ve serialization ayarları ilerideki Backend › Message Queue'larla Çalışmak serisinin konusu.

RabbitMQ: Exchange, Queue ve Binding

RabbitMQ'da producer bir queue'ya değil, bir exchange'e yazar. Exchange, message'ı binding kurallarına göre sıfır ya da daha fazla queue'ya kopyalar. Exchange'in türü kuralı belirler: direct (routing key birebir eşleşir), fanout (bağlı her queue'ya), topic (routing key'de joker karakterli desen eşleşmesi) ve headers. Publish/subscribe'ın RabbitMQ'daki karşılığı bu: her consumer grubu kendi queue'sunu açar ve exchange'e bağlar.

using System.Text.Json;
using Bookstore.Ordering.Contracts;
using RabbitMQ.Client;

namespace Bookstore.Messaging.RabbitMq;

// RabbitMQ: exchange'e gönder, routing key ile queue'lara dağıtılsın
public sealed class RabbitMqPublisher(IChannel channel, string exchange) : IMessagePublisher
{
    public async Task PublishAsync<T>(T message, string key, CancellationToken ct = default)
        where T : IIntegrationEvent
    {
        BasicProperties properties = new()
        {
            MessageId = Guid.CreateVersion7().ToString(),
            Type = T.MessageType,
            ContentType = "application/json",
            DeliveryMode = DeliveryModes.Persistent,
        };
        await channel.BasicPublishAsync(exchange, routingKey: T.MessageType, mandatory: false,
            basicProperties: properties, body: JsonSerializer.SerializeToUtf8Bytes(message), cancellationToken: ct);
    }
}

using System.Collections.Concurrent;
using System.Text.Json;
using Bookstore.Messaging.Kafka;
using Bookstore.Messaging.RabbitMq;
using Bookstore.Ordering;
using Bookstore.Ordering.Contracts;
using Confluent.Kafka;
using Confluent.Kafka.Admin;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

OrderPlaced[] orders =
[
    new(Guid.CreateVersion7(), "Ada", "Dune", 2, 60m),
    new(Guid.CreateVersion7(), "Alan", "Emma", 1, 20m),
    new(Guid.CreateVersion7(), "Grace", "The Hobbit", 1, 25m),
    new(Guid.CreateVersion7(), "Linus", "Middlemarch", 3, 66m),
    new(Guid.CreateVersion7(), "Ada", "Emma", 1, 20m),
    new(Guid.CreateVersion7(), "Alan", "Dune", 1, 30m),
];

// ---------- RabbitMQ: exchange + her consumer grubuna bir queue ----------
ConnectionFactory factory = new() { HostName = "localhost" };
await using IConnection connection = await factory.CreateConnectionAsync();
await using IChannel channel = await connection.CreateChannelAsync();

const string exchange = "bookstore.ordering";
await channel.ExchangeDeclareAsync(exchange, ExchangeType.Topic, durable: true);
Dictionary<string, object?> quorum = new() { ["x-queue-type"] = "quorum" };
foreach (string queue in new[] { "inventory.order-placed", "shipping.order-placed" })
{
    await channel.QueueDeclareAsync(queue, durable: true, exclusive: false, autoDelete: false, arguments: quorum);
    await channel.QueueBindAsync(queue, exchange, routingKey: OrderPlaced.MessageType);
}

ConcurrentDictionary<string, ConcurrentQueue<string>> received = new();
await StartRabbitConsumerAsync(connection, "inventory.order-placed", "inventory-1", received);
await StartRabbitConsumerAsync(connection, "shipping.order-placed", "shipping-1", received);
await StartRabbitConsumerAsync(connection, "shipping.order-placed", "shipping-2", received);

PlaceOrderHandler viaRabbit = new(new RabbitMqPublisher(channel, exchange));
foreach (OrderPlaced order in orders) await viaRabbit.HandleAsync(order);
await Task.Delay(TimeSpan.FromSeconds(2));

foreach ((string consumer, ConcurrentQueue<string> titles) in received.OrderBy(r => r.Key))
    Console.WriteLine($"[rabbitmq] {consumer,-12} got {titles.Count}: {string.Join(", ", titles)}");

// Yayından sonra açılan queue: geçmişi göremez
await channel.QueueDeclareAsync("analytics.order-placed", durable: true, exclusive: false, autoDelete: false, arguments: quorum);
await channel.QueueBindAsync("analytics.order-placed", exchange, routingKey: OrderPlaced.MessageType);
uint waiting = await channel.MessageCountAsync("analytics.order-placed");
Console.WriteLine($"[rabbitmq] analytics queue created after publishing: {waiting} messages");
Console.WriteLine();

static async Task StartRabbitConsumerAsync(
    IConnection connection, string queue, string name, ConcurrentDictionary<string, ConcurrentQueue<string>> received)
{
    IChannel channel = await connection.CreateChannelAsync();
    await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false);
    AsyncEventingBasicConsumer consumer = new(channel);
    consumer.ReceivedAsync += async (_, delivery) =>
    {
        OrderPlaced order = JsonSerializer.Deserialize<OrderPlaced>(delivery.Body.Span)!;
        received.GetOrAdd(name, _ => new()).Enqueue(order.Title);
        await Task.Delay(50);   // biraz iş yapıyormuş gibi
        await channel.BasicAckAsync(delivery.DeliveryTag, multiple: false);
    };
    await channel.BasicConsumeAsync(queue, autoAck: false, consumer);
}

Queue'lar quorum queue: RabbitMQ'nun Raft algoritmasıyla birden fazla node'a kopyalanan, kalıcı queue türü. RabbitMQ 4.0 ile eski classic mirrored queue'lar tamamen kaldırıldı; veri kaybetmemesi gereken queue'lar için önerilen tür artık bu. Çıktı:

[rabbitmq] inventory-1  got 6: Dune, Emma, The Hobbit, Middlemarch, Emma, Dune
[rabbitmq] shipping-1   got 3: Dune, The Hobbit, Dune
[rabbitmq] shipping-2   got 3: Emma, Middlemarch, Emma
[rabbitmq] analytics queue created after publishing: 0 messages

İlk soru cevaplandı. Inventory'nin tek consumer'ı altı message'ın hepsini aldı: her abone grubu kendi kopyasını alıyor. Shipping'in iki instance'ı aynı queue'dan okuduğu için message'ları paylaştı, her biri üçer tane aldı: competing consumers. prefetchCount: 1, bir consumer elindeki message'ı ack'lemeden ona yenisinin gönderilmemesini söylüyor; bu yüzden dağılım dengeli. Hangi siparişin hangi instance'a gittiği ise koşudan koşuya değişiyor; aynı programı ikinci kez çalıştırdığımda Shipping tarafındaki listeler farklıydı. Sıranın neden bozulduğu ve bunun ne zaman sorun olduğu serinin 05. yazısının konusu.

İkinci soru: yayından sonra açılan Analytics queue'su sıfır message görüyor. Exchange, message'ı yalnızca o anda bağlı olan queue'lara kopyalar; bağlı queue yoksa (mandatory flag ya da alternate exchange kurulmadıysa) message düşer. Ack'lenen message da queue'dan silinir. Queue-based broker'ın modeli bu: message bir iş birimidir, işlenince yok olur.

Kafka: Topic, Partition ve Consumer Group

Kafka'da channel bir topic, ama topic bir queue değil, bir log: sona eklenen, sırayla numaralanan ve okunduğunda silinmeyen kayıtlar. Topic, paralellik için partition'lara bölünür; her partition kendi içinde sıralı bir log'dur ve her kaydın partition içinde bir offset'i vardır. Producer'ın verdiği key, kaydın hangi partition'a gideceğini belirler.

using System.Text;
using System.Text.Json;
using Bookstore.Ordering.Contracts;
using Confluent.Kafka;

namespace Bookstore.Messaging.Kafka;

// Kafka: topic'e (log'a) ekle, key aynı partition'ı seçtirir
public sealed class KafkaPublisher(IProducer<string, byte[]> producer, string topic) : IMessagePublisher
{
    public async Task PublishAsync<T>(T message, string key, CancellationToken ct = default)
        where T : IIntegrationEvent
    {
        Message<string, byte[]> record = new()
        {
            Key = key,
            Value = JsonSerializer.SerializeToUtf8Bytes(message),
            Headers = new Headers
            {
                { "message-id", Encoding.UTF8.GetBytes(Guid.CreateVersion7().ToString()) },
                { "message-type", Encoding.UTF8.GetBytes(T.MessageType) },
            },
        };
        await producer.ProduceAsync(topic, record, ct);
    }
}

// ---------- Kafka: tek topic, 3 partition, consumer group'lar ----------
const string bootstrap = "localhost:9092";
const string topic = "bookstore.ordering";
using (IAdminClient admin = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrap }).Build())
{
    await admin.CreateTopicsAsync([new TopicSpecification { Name = topic, NumPartitions = 3, ReplicationFactor = 1 }]);
}

using IProducer<string, byte[]> producer =
    new ProducerBuilder<string, byte[]>(new ProducerConfig { BootstrapServers = bootstrap }).Build();
PlaceOrderHandler viaKafka = new(new KafkaPublisher(producer, topic));
foreach (OrderPlaced order in orders) await viaKafka.HandleAsync(order);

Console.WriteLine($"[kafka] shipping  (first read):  {ReadAll("shipping")}");
Console.WriteLine($"[kafka] analytics (new group):   {ReadAll("analytics")}");
Console.WriteLine($"[kafka] shipping  (second read): {ReadAll("shipping")}");

// Temizlik: örnek tekrar çalıştırılabilsin
foreach (string queue in new[] { "inventory.order-placed", "shipping.order-placed", "analytics.order-placed" })
    await channel.QueueDeleteAsync(queue);
await channel.ExchangeDeleteAsync(exchange);
using (IAdminClient admin = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrap }).Build())
{
    await admin.DeleteTopicsAsync([topic]);
    await admin.DeleteGroupsAsync(["shipping", "analytics"]);
}

string ReadAll(string groupId)
{
    ConsumerConfig config = new()
    {
        BootstrapServers = bootstrap,
        GroupId = groupId,
        AutoOffsetReset = AutoOffsetReset.Earliest,   // offset'i olmayan yeni grup baştan okur
    };
    using IConsumer<string, byte[]> consumer = new ConsumerBuilder<string, byte[]>(config).Build();
    consumer.Subscribe(topic);
    List<string> titles = [];
    DateTime until = DateTime.UtcNow.AddSeconds(8);
    while (DateTime.UtcNow < until)
    {
        ConsumeResult<string, byte[]>? result = consumer.Consume(TimeSpan.FromMilliseconds(500));
        if (result is null) { if (titles.Count > 0) break; continue; }
        titles.Add($"{JsonSerializer.Deserialize<OrderPlaced>(result.Message.Value)!.Title}@p{result.Partition.Value}");
    }
    consumer.Close();   // offset'ler commit edilir, gruptan çıkılır
    return titles.Count == 0 ? "nothing new" : $"{titles.Count} -> {string.Join(", ", titles)}";
}

[kafka] shipping  (first read):  6 -> Dune@p2, Emma@p1, The Hobbit@p1, Emma@p1, Middlemarch@p0, Dune@p0
[kafka] analytics (new group):   6 -> Dune@p2, Emma@p1, The Hobbit@p1, Emma@p1, Middlemarch@p0, Dune@p0
[kafka] shipping  (second read): nothing new

Okuma tarafının birimi consumer group. Aynı group id'yi kullanan consumer'lar topic'in partition'larını aralarında paylaşır: her partition, gruptaki tek bir consumer'a atanır. Bu da competing consumers, ama partition seviyesinde. Gruplar birbirinden bağımsızdır; her grup kaldığı yeri (offset) kendisi tutar. Çıktıdaki üç satır bunun sonuçları.

Shipping grubu altı kaydı okudu ve offset'lerini commit etti. Analytics grubu sonradan geldi ve aynı altı kaydı baştan okudu: AutoOffsetReset.Earliest, kayıtlı offset'i olmayan bir grubun log'un başından başlamasını sağlıyor. Shipping ikinci kez bağlandığında yeni bir şey yoktu, çünkü kaldığı yer log'un sonuydu. Üçüncü sorunun cevabı da bu: Kafka'da geç gelen abone geçmişi görür, yeter ki kayıtlar retention süresi içinde olsun. Topic'in varsayılan retention'ı 7 gün; bir topic için süresiz ya da key başına son kaydı tutan log compaction da seçilebilir.

Çıktıdaki sıraya da dikkat et: kayıtlar yayınlandıkları sırayla değil, partition partition gelmiş. Kafka sırayı yalnızca bir partition'ın içinde garanti eder. Bir başka sonuç: bir grupta partition sayısından fazla consumer çalıştırırsan fazlası boşta bekler; üç partition'lı bir topic'i en fazla üç consumer paralel okuyabilir. Kafka 4.2, bu sınırı esneten share group'ları (KIP-932, yani Kafka Improvement Proposal 932 ile gelen "Queues for Kafka") production'a hazır ilan etti: bir share group'taki consumer'lar partition'ları tek sahipli atamadan kayıtları birlikte tüketiyor, kayıt başına ack ve teslim denemesi sayımı yapılıyor. Yani Kafka da artık queue gibi davranabiliyor; bedeli, o kayıtlar için partition içi sıra garantisinden vazgeçmek.

Kavramdan Broker'a

Aynı EIP kavramının üç yaygın broker'daki karşılıkları:

Azure Service Bus'ın modeli RabbitMQ'ya Kafka'dan daha yakın: topic'e gelen her message, her subscription'a kopyalanır ve her subscription kendi başına bir queue gibi davranır. Fark, yönlendirmenin exchange ve binding yerine subscription'daki filter kurallarıyla yapılması.

Queue mu, Log mu?

İki model farklı soruların cevabı. Queue-based broker'da (RabbitMQ, Azure Service Bus) message bir iş birimi: birisi yapsın, yapıldığında unutulsun. Message bazında ack, yeniden deneme, gecikmeli teslim, öncelik ve dead-letter gibi özellikler bu modelde doğal. Log-based broker'da (Kafka) message bir olgu kaydı: kim isterse, ne zaman isterse okusun. Replay, yeni bir consumer'ı geçmişle beslemek, yüksek throughput ve aynı veriyi çok sayıda bağımsız grubun okuması bu modelde doğal.

Bookstore için okuma şu: "onay e-postası gönder" gibi command'lar ve tek seferlik işler queue'ya çok iyi oturuyor. OrderPlaced gibi birden çok context'in dinlediği ve ileride yeni context'lerin de dinlemek isteyeceği event'ler için log'un replay yeteneği değerli. Analytics ekibi altı ay sonra gelip "geçen çeyreğin siparişlerini de istiyoruz" dediğinde Kafka'da bu bir offset ayarıdır; RabbitMQ'da ayrı bir veri aktarımı projesi. Ama günde 800 sipariş alan, tek ops kişisi olan bir ekip için Kafka cluster'ı işletmek de ciddi bir yük. Sınırlar da bulanıklaşıyor: RabbitMQ'nun stream'leri (3.9'dan beri) replay edilebilen bir log sunuyor, Kafka'nın share group'ları da queue gibi tüketim. Seçimi broker'ın adına değil, ihtiyaç duyduğun davranışa göre yap: replay gerekiyor mu, message başına yeniden deneme ve gecikme gerekiyor mu, kaç bağımsız okuyucu olacak?

Soyutlama Ne Kadar İnce Olmalı?

IMessagePublisher iki broker'ı da saklıyor ve PlaceOrderHandler hangisiyle konuştuğunu bilmiyor. Ama soyutlama sızdırıyor ve bu dürüstçe kabul edilmeli. key parametresi RabbitMQ adapter'ında kullanılmıyor bile: orada yönlendirmeyi routing key yapıyor ve o da message tipinden geliyor. Kafka'da ise key, sıranın ve paralelliğin belirleyicisi. Uygulama kodu "sipariş id'si key'dir" kararını broker'dan bağımsız veriyor; bu kararın Kafka'daki sonucu serinin 05. yazısında.

İki adapter'ı ilk denemede aynı dosyaya yazdım ve derleyici durdurdu:

error CS0104: 'Headers' is an ambiguous reference between 'RabbitMQ.Client.Headers' and 'Confluent.Kafka.Headers'

Küçük bir hata, ama doğru bir sinyal: iki client kütüphanesi aynı dosyada, hatta aynı project'te olmamalı. Gerçek bir uygulamada her adapter kendi project'inde durur (Bookstore.Messaging.RabbitMq, Bookstore.Messaging.Kafka) ve composition root yalnızca birini kaydeder. Clean, Onion ve Hexagonal Architecture serisinde anlattığımız port/adapter ayrımının birebir uygulaması. Consumer tarafını bu kadar soyutlamaya çalışmak ise genellikle kötü fikir: ack, offset commit, prefetch ve partition ataması broker'a özgü davranışlar ve onları ortak bir interface'in arkasında aynıymış gibi göstermek hata yapmayı kolaylaştırır. MassTransit ve Wolverine gibi kütüphaneler bu işi hazır yapıyor; ayrıntıları Backend serisinde.

Sık Düşülen Tuzaklar

İki farklı consumer grubunu aynı queue'ya bağlamak

Shipping ekibi "OrderPlaced zaten inventory queue'sunda var" deyip oradan okumaya başlarsa, Inventory ile message'lar için yarışır: her biri siparişlerin yarısını görür. Competing consumers yalnızca aynı işi yapan instance'lar içindir. Farklı bir iş, farklı bir queue (ya da Kafka'da farklı bir consumer group) demek.

Yeni aboneyi geçmişle beslemeyi unutmak

RabbitMQ ve Azure Service Bus'ta yeni açılan queue ya da subscription yalnızca o andan sonraki message'ları görür. Yeni bir context'in mevcut siparişleri de bilmesi gerekiyorsa, bunun için ayrı bir başlangıç yükleme (backfill) planı yapılmalı; broker bunu kendiliğinden çözmez.

Partition'dan fazla consumer

Kampanya için Shipping consumer'ını Kafka'da on instance'a çıkarmak, topic'in üç partition'ı varsa yedi boşta instance demek. Klasik consumer group'ta paralelliğin üst sınırı partition sayısı; partition sayısını sonradan artırmak ise key'lerin partition'lara dağılımını değiştirir. Bunun sıra üzerindeki etkisi serinin 05. yazısında.

Bir command'ı publish/subscribe'a göndermek

Bir önceki yazıdaki uyarının channel karşılığı: ChargePayment bir fanout exchange'e gönderilirse, yarın ikinci bir queue bağlandığında para iki kez çekilir. Command'lar, tek bir queue'ya giden bir yola gönderilmeli.

Küçük Bir Deneme

★☆☆ Programda Shipping için üçüncü bir consumer başlat (shipping-3). Altı message nasıl dağılıyor? prefetchCount'u 10 yaparsan dağılım değişiyor mu, neden?

★★☆ RabbitMQ tarafında analytics.order-placed queue'sunu yayından önce oluştur ve bağla, ama ona consumer bağlama. Program bittiğinde queue'da kaç message var? Aynı şeyi Kafka'da yapmanın karşılığı nedir?

★★★ Kafka topic'ini 3 yerine 1 partition ile oluştur ve ReadAll'u iki ayrı consumer ile aynı group id üzerinden paralel çalıştır. İkinci consumer kaç kayıt alıyor? Sonra aynı denemeyi 3 partition ile tekrarla.

Bu yazıdaki consumer'lar hep mutlu yolda çalıştı: message geldi, işlendi, ack'lendi. Peki shipping-1 message'ı işledikten sonra ama ack'lemeden önce çökerse? Broker o message'ı ne yapar, Shipping aynı siparişi iki kez mi hazırlar? Bir sonraki yazıda delivery guarantee'lerine ve "exactly-once" vaadinin gerçekte neyi kapsadığına bakacağız.