RabbitMQ.Client 7 ile İlk Producer ve Consumer: Async API, Connection/Channel Yaşam Döngüsü ve Manual Ack

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

Billing consumer'ı billing.order-placed queue'sunu dinliyor ve her fatura 1 saniye sürüyor. Queue'da 500 mesaj bekliyor. Consumer'ı başlatıp 2,5 saniye sonra RabbitMQ'ya "queue'da kaç mesaj hazır bekliyor, kaçı consumer'da ack bekliyor?" diye soruyorsun. Kodda BasicQosAsync çağrısı yok. Cevap ne olur? Peki aynı consumer autoAck: true ile çalışıyor ve üçüncü mesajda process çöküyorsa, 20 mesajlık queue'da kaç mesaj kalır?

İki sorunun cevabı da yazının ortasında, gerçek rabbitmqctl çıktısıyla duruyor. Şimdiden söyleyeyim: ikisi de çoğu kişinin tahmininden kötü.

Connection ve Channel: Kim Ne Kadar Yaşar?

RabbitMQ.Client 7, Kasım 2024'te çıkan ve senkron API'yi (Application Programming Interface) tamamen kaldıran sürüm. Artık her şey async: CreateConnectionAsync, CreateChannelAsync, BasicPublishAsync. 6.x'ten geçişin ayrıntıları serinin son yazısında; burada sıfırdan doğru kodu yazıyoruz. İki temel object var. IConnection broker'a açılan TCP (Transmission Control Protocol) bağlantısını temsil eder; açılması TCP el sıkışması, AMQP (Advanced Message Queuing Protocol) el sıkışması ve kimlik doğrulama demek. IChannel bu bağlantının içinde açılan hafif bir mantıksal oturumdur; publish, consume, ack gibi bütün işlemler channel üzerinden yapılır.

Kural basit: connection uygulama ömrü boyunca tek ve uzun ömürlü, channel ise iş başına ayrı ve paylaşılmayan. RabbitMQ'nun .NET client rehberi bir IChannel'ın aynı anda birden fazla thread tarafından kullanılmamasını, özellikle publish eden thread'ler arasında hiç paylaşılmamasını söylüyor; aynı channel'a eşzamanlı publish, protokol frame'lerinin birbirine karışmasına yol açar. Order API'nin Program.cs'i bu yüzden şöyle:

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

WebApplicationBuilder builder = WebApplication.CreateBuilder(args);

ConnectionFactory factory = new()
{
    Uri = new Uri(builder.Configuration.GetConnectionString("RabbitMq")!),
    ClientProvidedName = "bookstore-order-api"
};

// Uygulama ömrü boyunca tek connection: startup'ta bir kez açılır
IConnection connection = await factory.CreateConnectionAsync();
await RabbitTopologySetup.DeclareAsync(connection);

builder.Services.AddSingleton(connection);
builder.Services.AddSingleton<OrderEventPublisher>();

WebApplication app = builder.Build();

app.MapPost("/orders", async (PlaceOrderRequest request, OrderEventPublisher publisher, CancellationToken ct) =>
{
    OrderPlaced message = new(
        OrderId: Guid.CreateVersion7(),
        CustomerId: request.CustomerId,
        Total: request.Items.Sum(i => i.Quantity * i.UnitPrice),
        Currency: "TRY",
        PlacedAt: DateTimeOffset.UtcNow,
        Items: request.Items);

    await publisher.PublishAsync(message, ct);
    return Results.Accepted($"/orders/{message.OrderId}", new { message.OrderId });
});

app.Lifetime.ApplicationStopping.Register(() => app.Logger.LogInformation("Stopping, connection will be closed"));

await app.RunAsync();
await connection.CloseAsync();

public sealed record PlaceOrderRequest(Guid CustomerId, IReadOnlyList<OrderPlacedItem> Items);

Connection'ı DI (Dependency Injection) container'ına bir factory lambda'sı ile değil, top-level await ile oluşturulmuş hazır bir instance olarak veriyoruz. Lambda içinde CreateConnectionAsync().GetAwaiter().GetResult() yazmak hem senkron bekleme hem de "ilk istekte bağlantı açılıyor" sürprizi demek. Burada broker'a ulaşılamıyorsa uygulama hiç ayağa kalkmaz; bu da istediğimiz şey, çünkü RabbitMQ'nun automatic recovery'si ilk bağlantı denemesini kapsamıyor. Lifetime'ların ayrıntısı ASP.NET Core Web API serisinin "Dependency Injection: Lifetime'lar ve Captive Dependency Tuzağı" yazısında. ClientProvidedName küçük ama değerli bir ayar: management arayüzünde ve rabbitmqctl list_connections çıktısında bağlantıyı hangi uygulamanın açtığını gösterir.

"Her mesaja yeni connection" ne kadar pahalı? Aynı makinede, aynı broker'a 200 mesaj gönderen küçük bir deneme:

New connection per message: 200 messages in 393 ms
One long-lived connection: 200 messages in 1 ms
Long-lived + confirms, awaited one by one: 200 messages in 38 ms
Long-lived + confirms, batch awaited: 200 messages in 3 ms

Broker localhost'ta olduğu hâlde her mesaj için bağlantı açmak yaklaşık 400 kat yavaş. Gerçek ağda, TLS (Transport Layer Security) ile bu fark daha da büyür ve broker tarafında her bağlantı ayrı bir Erlang process'i ve bellek demektir. Son iki satır publisher confirmation'ın maliyetini gösteriyor; ona birazdan geleceğiz.

Topology'yi Startup'ta Declare Etmek

Exchange ve queue'ları elle management arayüzünden oluşturmak yerine uygulama başlarken declare ediyoruz:

public static class RabbitTopologySetup
{
    public static async Task DeclareAsync(IConnection connection)
    {
        await using IChannel channel = await connection.CreateChannelAsync();

        await channel.ExchangeDeclareAsync(RabbitTopology.OrderingExchange, ExchangeType.Topic, durable: true);
        await channel.QueueDeclareAsync(RabbitTopology.BillingQueue, durable: true, exclusive: false, autoDelete: false,
            arguments: new Dictionary<string, object?> { ["x-queue-type"] = "quorum" });
        await channel.QueueBindAsync(RabbitTopology.BillingQueue, RabbitTopology.OrderingExchange,
            RabbitTopology.OrderPlacedRoutingKey);
    }
}

Declare işlemleri idempotent: aynı isim ve aynı argümanlarla ikinci kez çağırmak hiçbir şey değiştirmez. bookstore.ordering bir topic exchange, billing.order-placed bir quorum queue ve aradaki binding order.placed routing key'i ile kuruluyor. Exchange tiplerini, quorum queue'yu ve hangi ayarın kodda, hangisinin broker policy'sinde durması gerektiğini bir sonraki yazıda konuşacağız.

Publish: BasicPublishAsync ve Publisher Confirmation

Publisher, singleton olarak yaşayan ve tek bir channel'ı bir SemaphoreSlim arkasında kullanan küçük bir class:

public sealed class OrderEventPublisher(IConnection connection, ILogger<OrderEventPublisher> logger) : IAsyncDisposable
{
    private readonly SemaphoreSlim _gate = new(1, 1);
    private IChannel? _channel;

    public async Task PublishAsync(OrderPlaced message, CancellationToken ct)
    {
        byte[] body = JsonSerializer.SerializeToUtf8Bytes(message, MessageJson.Options);
        BasicProperties props = new()
        {
            MessageId = Guid.NewGuid().ToString(),
            CorrelationId = message.OrderId.ToString(),
            Type = OrderPlaced.MessageType,
            ContentType = "application/json",
            DeliveryMode = DeliveryModes.Persistent
        };

        // IChannel thread-safe değil: aynı anda tek publish
        await _gate.WaitAsync(ct);
        try
        {
            _channel ??= await connection.CreateChannelAsync(new CreateChannelOptions(
                publisherConfirmationsEnabled: true,
                publisherConfirmationTrackingEnabled: true), ct);

            // confirm tracking açıkken bu await broker "ack" diyene kadar sürer
            await _channel.BasicPublishAsync(RabbitTopology.OrderingExchange, RabbitTopology.OrderPlacedRoutingKey,
                mandatory: true, basicProperties: props, body: body, cancellationToken: ct);
        }
        finally
        {
            _gate.Release();
        }

        logger.LogInformation("Published {MessageType} {MessageId} for order {OrderId}",
            props.Type, props.MessageId, message.OrderId);
    }

    public async ValueTask DisposeAsync()
    {
        if (_channel is not null)
        {
            await _channel.CloseAsync();
            await _channel.DisposeAsync();
        }
    }
}

Burada üç karar var. Birincisi CreateChannelOptions: 7.x'te publisher confirmation channel açılırken açılıyor. publisherConfirmationsEnabled broker'dan her mesaj için onay istemek, publisherConfirmationTrackingEnabled ise bu onayı client'ın takip edip BasicPublishAsync'in dönen ValueTask'ına bağlaması demek. İkisi açıkken await broker mesajı aldığını onaylayana kadar bitmez; broker reddederse exception fırlar. Confirm olmadan BasicPublishAsync'in başarıyla dönmesi sadece "byte'lar socket'e yazıldı" demektir.

İkincisi mandatory: true. Exchange mesajı hiçbir queue'ya yönlendiremezse, örneğin binding'i olmayan bir routing key kullandıysan, broker mesajı varsayılan olarak sessizce atar. mandatory ile confirm tracking birlikte açıkken aynı hata publish edene exception olarak döner. Binding'i olmayan order.refunded routing key'i ile denediğimizde gerçek çıktı:

PublishReturnException: IsReturn=True, SeqNo=1, 312 NO_ROUTE Exchange: bookstore.ordering Routing Key: order.refunded
Without confirms and mandatory: publish 'succeeded', message silently dropped

Üçüncüsü mesaj property'leri. MessageId her mesaja bir kez verilir ve redelivery'lerde değişmez; ilerideki idempotent consumer yazısında bu id'yi dedupe için kullanacağız. CorrelationId siparişi, Type contract'ın adını ve sürümünü taşır. DeliveryMode = Persistent mesajı diske yazdırır; quorum queue'larda mesajlar zaten her zaman diske yazılıyor, ama classic queue'ya düşen mesaj için bu ayar fark yaratır.

Confirm'ün maliyetine dönelim. Yukarıdaki ölçümde 200 mesajı tek tek await etmek 38 ms, aynı 200 publish'in ValueTask'larını bir listede toplayıp sonra beklemek 3 ms sürdü. Her await bir broker tur yolculuğu bekler. Order API'de her HTTP (Hypertext Transfer Protocol) isteği tek mesaj ürettiği için tek tek beklemek doğru: istemciye 202 dönmeden mesajın broker'da olduğundan emin oluyoruz. Toplu import gibi senaryolarda ise batch halinde bekle.

Gerçek koşuda POST /orders isteği ve API logu:

HTTP/1.1 202 Accepted
Location: /orders/01a1b8de-6a5a-70fd-8aab-7f9dd66cdbe7

{"orderId":"01a1b8de-6a5a-70fd-8aab-7f9dd66cdbe7"}

info: Bookstore.OrderApi.OrderEventPublisher[0]
      Published bookstore.order-placed.v1 dfa727ef-8eae-473e-a4c7-5f89a484da6d for order 01a1b8de-6a5a-70fd-8aab-7f9dd66cdbe7

Consumer'ı BackgroundService İçinde Çalıştırmak

Billing tarafı bir Worker Service. Consumer, BackgroundService'in ExecuteAsync'i içinde kendi channel'ını açıyor:

public sealed class OrderPlacedConsumer(
    IConnection connection, InvoiceService invoices, ILogger<OrderPlacedConsumer> logger) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await using IChannel channel = await connection.CreateChannelAsync(cancellationToken: stoppingToken);

        // ack'lenmemiş en fazla 10 mesaj bu consumer'a gelir
        await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 10, global: false, stoppingToken);

        AsyncEventingBasicConsumer consumer = new(channel);
        consumer.ReceivedAsync += async (_, ea) =>
        {
            try
            {
                OrderPlaced order = JsonSerializer.Deserialize<OrderPlaced>(ea.Body.Span, MessageJson.Options)
                    ?? throw new JsonException("Empty body");

                await invoices.CreateInvoiceAsync(order, stoppingToken);
                await channel.BasicAckAsync(ea.DeliveryTag, multiple: false);
            }
            catch (JsonException ex)
            {
                // bozuk mesaj tekrar denemekle düzelmez: requeue etme
                logger.LogError(ex, "Malformed message {MessageId}, rejecting", ea.BasicProperties.MessageId);
                await channel.BasicNackAsync(ea.DeliveryTag, multiple: false, requeue: false);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                // kapanıyoruz: ack yok, broker mesajı başka consumer'a verir
                logger.LogWarning("Shutdown during {MessageId}, leaving it unacked", ea.BasicProperties.MessageId);
            }
        };

        string consumerTag = await channel.BasicConsumeAsync(RabbitTopology.BillingQueue, autoAck: false,
            consumer: consumer, cancellationToken: stoppingToken);
        logger.LogInformation("Consuming {Queue} with tag {ConsumerTag}", RabbitTopology.BillingQueue, consumerTag);

        try
        {
            await Task.Delay(Timeout.Infinite, stoppingToken);
        }
        catch (OperationCanceledException)
        {
            // normal kapanış
        }

        await channel.BasicCancelAsync(consumerTag, noWait: false, CancellationToken.None);
        await channel.CloseAsync(CancellationToken.None);
        logger.LogInformation("Consumer {ConsumerTag} cancelled", consumerTag);
    }
}

AsyncEventingBasicConsumer 7.x'te tek consumer tipi gibi düşünülebilir; event'in adı ReceivedAsync ve handler bir Task döner. Mesaj gövdesi ReadOnlyMemory<byte> olarak gelir ve client bu belleği handler bittikten sonra yeniden kullanabilir. Bu yüzden gövdeyi handler içinde deserialize ediyoruz; ea.Body'yi bir field'a saklayıp sonra okumak bozuk veri okumak demek.

Varsayılan olarak bir channel'daki mesajlar sırayla, tek tek dispatch edilir: ConsumerDispatchConcurrency 1. Handler 200 ms sürüyorsa bu consumer saniyede en fazla 5 mesaj işler. Paralellik istiyorsan ya ConsumerDispatchConcurrency'yi artırırsın ya da aynı queue'ya birden fazla consumer (birden fazla instance veya channel) bağlarsın. İkinci yol hem daha basit hem de sıralama beklentisini daha görünür kılar.

ExecuteAsync'in neden "consume başlat ve sonsuza kadar bekle" şeklinde olduğu da önemli. Consume işlemi bir döngü değil, broker'ın push ettiği bir abonelik; BasicConsumeAsync hemen döner. Method biterse await using channel'ı kapatır ve abonelik ölür. .NET 10'da ExecuteAsync'in tamamı arka plandaki bir thread'de çalışıyor, yani await'ten önceki kısmı artık host'un açılışını bloklamıyor; loglarda Consuming billing.order-placed satırının Application started'dan sonra gelmesinin sebebi bu. BackgroundService'in yaşam döngüsünü ilerideki Background Job'lar serisinde ayrıntılı konuşacağız.

Manual Ack, Nack ve Prefetch

Yazının başındaki ilk soruya dönelim. Prefetch yoksa broker, consumer'a queue'daki her mesajı gönderir; hepsi client'ın belleğinde, ack bekleyerek durur. 500 mesaj, her biri 2 KB, handler 1 saniye; consumer başladıktan 2,5 saniye sonra:

prefetch=none: consumer started, check the queue now
name	messages_ready	messages_unacknowledged	consumers
lab.prefetch	0	498	1
prefetch=10: consumer started, check the queue now
name	messages_ready	messages_unacknowledged	consumers
lab.prefetch	488	10	1

Prefetch'siz consumer iki mesajı bitirmiş, kalan 498'in tamamını kendi belleğine çekmiş. Queue'da "hazır" mesaj sıfır görünüyor. Bu sırada ikinci bir Billing instance'ı başlatsan ona tek mesaj düşmez; hepsi ilk instance'ta kilitli. Mesajlar 2 KB değil 200 KB olsaydı process'in belleği de buna göre şişerdi. BasicQosAsync(0, 10, false) ile aynı deneme: consumer'da 10 mesaj, queue'da 488 mesaj bekliyor ve yeni bir instance hemen pay alabilir.

Doğru prefetch değeri handler süresine bağlı. Hızlı ve CPU'ya (Central Processing Unit) bağlı handler'larda 50–300 arası bir değer ağ gecikmesini gizler; Billing gibi yavaş, dışarıya çağrı yapan handler'larda 5–20 makul bir başlangıç. Sabit bir doğru yok; ölç ve ayarla.

İkinci soru: autoAck: true. Bu modda broker mesajı consumer'a yazdığı anda teslim edilmiş sayar ve queue'dan siler. Consumer'ın işi bitirip bitirmediği umurunda değildir. Prefetch de devreye girmez, çünkü ack beklenen mesaj yoktur. 20 mesajlık queue'yu autoAck: true ile tüketen ve üçüncü mesajda çöken consumer:

20 messages published to lab.autoack
Processing invoice-1
Processing invoice-2
Processing invoice-3
Process crashed
exit=1
lab.autoack	0	0

Üç mesaj işlenmeye başlandı, ama queue'da kalan mesaj sıfır. Broker 20 mesajın hepsini ilk anda client'a basmıştı; process ölünce 17'si hiç işlenmeden yok oldu. Faturası kesilmeyen 17 sipariş, hiçbir hata kaydı olmadan. autoAck: false ile aynı çöküşte ack'lenmemiş her mesaj queue'ya geri döner. Bu, Messaging ve Event-Driven Architecture serisinin "Delivery Guarantee'leri" yazısında anlatılan at-most-once ile at-least-once arasındaki farkın kodda tam karşılığı.

Ack'in yanında iki olumsuz cevap var. BasicNackAsync(tag, multiple: false, requeue: true) mesajı queue'ya geri koyar; requeue: false ise mesajı atar ya da queue'nun dead-letter ayarı varsa DLX'e (Dead Letter Exchange) gönderir. Bozuk JSON (JavaScript Object Notation) tekrar denemekle düzelmeyeceği için yukarıdaki kodda requeue: false kullandık. Retry ve dead-letter stratejileri serinin ilerideki iki yazısının konusu.

Bağlantı Koptuğunda: Automatic Recovery

7.x'te ConnectionFactory'nin varsayılanları, gerçek koşudan:

AutomaticRecoveryEnabled = True
TopologyRecoveryEnabled  = True
NetworkRecoveryInterval  = 00:00:05
RequestedHeartbeat       = 00:01:00
ConsumerDispatchConcurrency = 1

Automatic recovery açıkken bağlantı koptuğunda client 5 saniyede bir yeniden bağlanmayı dener; başarılı olunca channel'ları, QoS (Quality of Service) ayarlarını, client'ın declare ettiği exchange, queue ve binding'leri ve consumer'ları yeniden kurar. Billing ve Order API çalışırken RabbitMQ container'ını durdurup yeniden başlattık. Billing logu:

Connection shutdown: 320 CONNECTION_FORCED - broker forced connection closure with reason 'shutdown'
Connection recovered
info: Bookstore.Billing.InvoiceService[0]
      Invoice created for order 01a1b8de-bc82-73c1-b8cd-ffa17698f366: 450.00 TRY

Consumer hiçbir kod yazmadan geri geldi ve bir sonraki siparişi işledi. Publisher tarafı ise daha az şanslı. Broker kapalıyken gelen istek:

fail: Microsoft.AspNetCore.Diagnostics.DeveloperExceptionPageMiddleware[1]
      An unhandled exception has occurred while executing the request.
      RabbitMQ.Client.Exceptions.AlreadyClosedException: Already closed: The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=320, text='CONNECTION_FORCED - broker forced connection closure with reason 'shutdown'', classId=0, methodId=0

Client, recovery sırasında publish edilen mesajları tamponlamıyor; exception fırlatıyor. Bu dürüst bir davranış: API 500 döndü ve sipariş "alındı" denmedi. Ama sipariş veritabanına yazılıp sonra publish düşerse elinde tutarsız bir durum kalır. Bunun gerçek çözümü transactional outbox ve serinin ilerideki outbox yazısında .NET tarafını kuracağız. Recovery sonrası bir başka ince nokta: client eski channel'dan kalan delivery tag'leri ack'lemeye çalışmaz, yani recovery anında işlenmekte olan mesajlar yeniden teslim edilir. Consumer'ın aynı mesajı iki kez görebileceğini baştan kabul et.

Graceful Shutdown

Deploy sırasında Billing'e SIGTERM (Unix'te process'e "kapan" diyen sonlandırma sinyali) gelir. Host stoppingToken'ı iptal eder, Task.Delay biter, BasicCancelAsync broker'a "bana yeni mesaj gönderme" der ve channel kapanır. Beş sipariş gönderip 300 ms sonra process'e SIGTERM yolladık:

info: Microsoft.Hosting.Lifetime[0]
      Application is shutting down...
warn: Bookstore.Billing.OrderPlacedConsumer[0]
      Shutdown during 0a024092-1151-492b-86eb-fc979a82305a, leaving it unacked
warn: Bookstore.Billing.OrderPlacedConsumer[0]
      Shutdown during d44eaba8-d344-4ce3-b79b-320cdb159a89, leaving it unacked
warn: Bookstore.Billing.OrderPlacedConsumer[0]
      Shutdown during b7cfdfb2-014a-4156-91b8-df0f5dd4300e, leaving it unacked
warn: Bookstore.Billing.OrderPlacedConsumer[0]
      Shutdown during 0a2950b8-e18f-4291-aa89-2de6a3d2e610, leaving it unacked
info: Bookstore.Billing.OrderPlacedConsumer[0]
      Consumer amq.ctag-I7Fm7V6xSAGRCy5Tokf_BQ cancelled
Connection shutdown: 200 Goodbye

name	messages_ready	messages_unacknowledged	messages
billing.order-placed	4	0	4

Bir fatura kapanmadan önce bitmişti; prefetch ile client'a gelmiş diğer dört mesaj iptal edilen token yüzünden yarıda kaldı, ack'lenmedi ve channel kapanınca queue'ya geri döndü. Kayıp yok. Burada bilinçli bir seçim yaptık: handler'a stoppingToken verdik, yani kapanışta yarım iş iptal ediliyor. Alternatif, handler'a CancellationToken.None verip yarım işin bitmesini beklemek. O durumda host'un shutdown timeout'u (varsayılan 30 saniye) içinde bitmesi gerekir; prefetch büyükse bu süre yetmeyebilir. İptalin veritabanına kadar taşınması ASP.NET Core Web API serisinin "Async/Await ve CancellationToken" yazısında.

Üretimde Dikkat

Bağlantı sayısını izle. Management arayüzünde bağlantı sayısı instance sayısıyla orantılı ve sabit olmalı; istek sayısıyla birlikte artıyorsa bir yerde connection sızıyordur. ClientProvidedName bu aramayı dakikalara indirir.

Channel'ları da izle. RabbitMQ 4.x'te bağlantı başına channel sayısının varsayılan üst sınırı (channel_max) 2047. İstek başına channel açıp kapatmayı unutan kod bu sınıra dayanır ve yeni channel açılamaz.

Queue'nun iki sayısını ayrı ayrı izle: messages_ready tüketilmeyi bekleyen birikmeyi, messages_unacknowledged consumer'larda takılı kalan işi gösterir. Unacked sayısı sürekli prefetch × consumer sayısında duruyorsa consumer'lar yetişemiyor demektir. Bu metric'leri OpenTelemetry ile toplamayı serinin observability yazısında yapacağız.

Publish yolunda timeout koy. Confirm bekleyen bir await, broker bellek veya disk alarmındayken (publisher'ları bloklar) uzun süre dönmeyebilir. HTTP isteğinin CancellationToken'ını ya da kendi timeout'unu BasicPublishAsync'e ver.

Tek channel'lı publisher'ın verimi bir broker tur yolculuğuyla sınırlı. Order API saniyede yüzlerce sipariş almıyorsa bu yeterli; alıyorsa birkaç channel'lık küçük bir havuz ya da batch confirm düşün.

Sık Düşülen Tuzaklar

Her mesaj için yeni connection. Ölçtüğümüz 400 katlık fark localhost'ta; gerçek ortamda broker'ı da yorar. Connection singleton, channel iş başına.

Channel'ı thread'ler arasında paylaşmak. Singleton bir publisher içinde tek IChannel'ı kilitsiz kullanmak, yük altında ara sıra görülen protokol hatalarına yol açar; bunlar test ortamında nadiren görünür.

autoAck: true ile "bir şey olmaz" demek. Gördüğümüz gibi process çöktüğünde prefetch sınırı olmadan client'a basılmış bütün mesajlar kaybolur.

Prefetch'siz consumer. Queue'nun tamamı tek instance'ın belleğine akar, yatay ölçekleme işe yaramaz hâle gelir.

Confirm'süz publish'e güvenmek. BasicPublishAsync'in exception fırlatmaması, mesajın bir queue'ya ulaştığı anlamına gelmez; yanlış routing key ile gönderilen mesaj sessizce yok olur.

ea.Body'yi handler dışına taşımak. Bellek client'a aittir; kopyala ya da handler içinde deserialize et.

Küçük Bir Deneme

★☆☆ Billing'i iki instance olarak çalıştır ve 20 sipariş gönder. Prefetch 10 iken mesajlar iki instance arasında nasıl dağılıyor? Prefetch'i 1'e indirip tekrar dene.

★★☆ ConsumerDispatchConcurrency'yi 4 yap ve handler içinde hangi thread'de çalıştığını logla. 20 siparişin işlenme sırası değişiyor mu? Sıralı işlem gereken bir senaryoda bu ayarın riskini bir paragrafla yaz.

★★★ OrderEventPublisher'ı tek channel yerine 4 channel'lık bir havuzla yeniden yaz. 1.000 eşzamanlı isteği bir yük testi aracıyla gönder ve tek channel'lı sürümle p95 gecikmesini karşılaştır.

Şimdilik tek bir topic exchange, tek bir binding ve tek bir quorum queue ile çalıştık. Bir sonraki yazıda topology'yi kodda ciddiye alıyoruz: direct, topic, fanout ve headers exchange'leri, routing key tasarımı, classic, quorum ve stream queue'lar arasındaki fark ve hangi ayarın client'ta değil broker policy'sinde durması gerektiği.