Azure Service Bus .NET SDK: Queue, Topic/Subscription, Session'lar ve ServiceBusProcessor

Sürüm notu: Bu yazıdaki kodlar .NET 10, Azure.Messaging.ServiceBus 7.21.0 ve Microsoft.Extensions.Azure 1.14.1 ile derlenip Azure Service Bus emulator'ü üzerinde denenmiştir.

Önce Tahmin Et

orders queue'sunun lock süresi 30 saniye. Bir ServiceBusProcessor mesajı alıyor, handler 40 saniye çalışıyor ve sonunda CompleteMessageAsync çağırıyor. Processor'da MaxAutoLockRenewalDuration = TimeSpan.Zero var. CompleteMessageAsync başarılı olur mu? Olmazsa mesaj nerede, ve bir sonraki teslimde DeliveryCount kaç olur?

İkinci soru: aynı queue'nun MaxDeliveryCount değeri 3. Handler her mesajda exception fırlatıyor ve mesajı ne complete ne abandon ediyor. Mesajın dead-letter queue'ya düşmesi kaç saniye sürer: 90 saniye mi (3 × lock süresi), yoksa daha az mı?

ServiceBusClient'ı Singleton Tutmak ve Kimlik

RabbitMQ'da connection'ı uzun ömürlü tutup channel'ları iş başına açıyorduk. Azure Service Bus SDK'sında (Software Development Kit) karşılığı ServiceBusClient: AMQP (Advanced Message Queuing Protocol) bağlantısını yönetir, thread-safe'tir ve uygulama ömrü boyunca tek instance olarak yaşamalıdır. Ondan üretilen ServiceBusSender, ServiceBusReceiver ve ServiceBusProcessor da thread-safe ve cache'lenebilir. Kaydı Microsoft.Extensions.Azure paketindeki AddAzureClients ile yapıyoruz:

using Azure.Identity;
using Bookstore.Billing.ServiceBus;
using Microsoft.Extensions.Azure;

HostApplicationBuilder builder = Host.CreateApplicationBuilder(args);

builder.Services.AddAzureClients(clients =>
{
    string? connectionString = builder.Configuration.GetConnectionString("ServiceBus");
    if (!string.IsNullOrEmpty(connectionString))
    {
        // yerel geliştirme: emulator connection string'i
        clients.AddServiceBusClient(connectionString);
    }
    else
    {
        // Azure: secret yok, managed identity veya geliştiricinin Azure CLI oturumu
        clients.AddServiceBusClientWithNamespace(builder.Configuration["ServiceBus:Namespace"]!);
        clients.UseCredential(new DefaultAzureCredential());
    }
});

builder.Services.AddSingleton<InvoiceService>();
builder.Services.AddHostedService<OrderPlacedProcessor>();

await builder.Build().RunAsync();

AddAzureClients, ServiceBusClient'ı singleton olarak kaydeder ve container kapanırken dispose eder. Azure'da connection string yerine Microsoft Entra ID ile kimlik doğrulama kullan: namespace adı (bookstore.servicebus.windows.net gibi) bir secret değildir, uygulamanın managed identity'sine "Azure Service Bus Data Sender" ve "Azure Service Bus Data Receiver" rolleri verilir. DefaultAzureCredential geliştirme makinesinde Azure CLI (Command Line Interface) oturumunu, Azure'da managed identity'yi bulur. Bir not: Azure Identity dokümantasyonu production'da credential zincirini tahmin edilebilir kılmak için doğrudan ManagedIdentityCredential gibi belirli bir credential kullanmayı da öneriyor; zincirin her adımı denenirken yaşanan gecikmeler başlangıçta can sıkabilir.

Connection string'de RootManageSharedAccessKey görüyorsan dikkat: bu, namespace üzerinde her şeyi yapabilen bir anahtar. Emulator'de sorun değil; Azure'da uygulamaya asla bu anahtarı verme.

Yerelde Service Bus Emulator

Microsoft'un resmi Service Bus emulator'ü bir Linux container'ı ve verisini bir SQL (Structured Query Language) Server veritabanında tutuyor. Bu yazıdaki her çıktı bu emulator'den. Kurulum iki container:

docker network create mqwork1-net
docker run -d --name mqwork1-sql --network mqwork1-net -e ACCEPT_EULA=Y -e MSSQL_SA_PASSWORD="$SQL_PASSWORD" mcr.microsoft.com/mssql/server:2022-latest
docker run -d --name mqwork1-sbemu --network mqwork1-net -p 17220:5672 -p 17221:5300 -e SQL_SERVER=mqwork1-sql -e MSSQL_SA_PASSWORD="$SQL_PASSWORD" -e ACCEPT_EULA=Y -v "$PWD/sbemu/Config.json:/ServiceBus_Emulator/ConfigFiles/Config.json" mcr.microsoft.com/azure-messaging/servicebus-emulator:latest

Queue, topic ve subscription'lar Config.json'da tanımlanır. Bookstore için orders queue'su, session'lı order-status queue'su ve billing, notification, audit subscription'larına sahip bookstore-orders topic'i var. Billing subscription'ının kısaltılmış tanımı:

{
  "Name": "billing",
  "Properties": {
    "LockDuration": "PT30S",
    "MaxDeliveryCount": 3,
    "RequiresSession": false
  },
  "Rules": [
    {
      "Name": "order-placed-only",
      "Properties": {
        "FilterType": "Correlation",
        "CorrelationFilter": { "Label": "OrderPlaced" }
      }
    }
  ]
}

Emulator açıldığında logda bağlantı cümlesini basıyor; port'u değiştirdiysek endpoint'e eklememiz yeterli: Endpoint=sb://localhost:17220;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;. Emulator'ün sınırlarını bil: container yeniden başlayınca entity'ler ve mesajlar kalıcı değil, partitioned entity desteklenmiyor, mesaj boyutu en fazla 256 KB, namespace başına en fazla 10 eşzamanlı bağlantı ve yalnızca AMQP over TCP (Transmission Control Protocol) var, WebSocket yok. Microsoft'un deyişiyle emulator sıralı geliştirme ve test senaryoları için; yük testini emulator'de yapma.

Gönderim, Topic ve Subscription Filter'ları

Mesajı kurarken RabbitMQ'daki property'lerin karşılıklarını dolduruyoruz:

static ServiceBusMessage ToMessage(OrderPlaced order) => new(BinaryData.FromObjectAsJson(order, MessageJson.Options))
{
    MessageId = Guid.NewGuid().ToString(),
    CorrelationId = order.OrderId.ToString(),
    Subject = "OrderPlaced",
    ContentType = "application/json",
    ApplicationProperties = { ["type"] = OrderPlaced.MessageType, ["total"] = (double)order.Total }
};

await using ServiceBusSender sender = client.CreateSender("bookstore-orders");
await sender.SendMessageAsync(ToMessage(NewOrder(1220.50m)));

// tek tek göndermek yerine batch: boyut sınırını SDK hesaplar
using ServiceBusMessageBatch batch = await sender.CreateMessageBatchAsync();
foreach (decimal total in new[] { 450m, 89.90m, 2400m })
{
    if (!batch.TryAddMessage(ToMessage(NewOrder(total))))
    {
        throw new InvalidOperationException("Message too large for the batch");
    }
}
await sender.SendMessagesAsync(batch);

SendMessageAsync broker mesajı kalıcı olarak kabul ettiğinde tamamlanır; RabbitMQ'da publisher confirmation için ayrıca yaptığımız işi burada SDK kendisi yapıyor. CreateMessageBatchAsync boyut sınırını entity'den okuyor ve TryAddMessage sınır aşılınca false dönüyor. Gerçek çıktı:

Batch: 3 messages, 1446 bytes, max 262144 bytes
Sent 4 OrderPlaced + 1 OrderCancelled

Topic'e dört OrderPlaced ve bir OrderCancelled gönderdik. Her subscription topic'in kendi kopyasını tutan bir sanal queue'dur ve hangi mesajları alacağını rule'ları belirler. Billing ve Notification'da Subject = 'OrderPlaced' correlation filter'ı, Audit'te ise total > 1000 SQL filter'ı var. Subscription'lara peek ile baktığımızda:

billing      4 messages: OrderPlaced/1220.5, OrderPlaced/450, OrderPlaced/89.9, OrderPlaced/2400
notification 4 messages: OrderPlaced/1220.5, OrderPlaced/450, OrderPlaced/89.9, OrderPlaced/2400
audit        2 messages: OrderPlaced/1220.5, OrderPlaced/2400

OrderCancelled hiçbir subscription'a girmedi, Audit yalnızca 1.000 TL'nin üstündeki iki siparişi aldı. Filter'lar body'ye bakamaz; yalnızca sistem property'lerine (Subject, CorrelationId…) ve ApplicationProperties'e bakar. total'ı bu yüzden property olarak da taşıyoruz. Mümkünse correlation filter kullan: Microsoft'un rehberine göre SQL filter'lardan daha verimli değerlendiriliyor, emulator'de de topic başına SQL filter kotası (70) correlation filter kotasından (1.000) çok daha düşük.

ServiceBusProcessor ile Tüketmek

Billing, billing subscription'ını bir BackgroundService içindeki ServiceBusProcessor ile tüketiyor:

public sealed class OrderPlacedProcessor(
    ServiceBusClient client, InvoiceService invoices, ILogger<OrderPlacedProcessor> logger) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await using ServiceBusProcessor processor = client.CreateProcessor("bookstore-orders", "billing",
            new ServiceBusProcessorOptions
            {
                AutoCompleteMessages = false,
                MaxConcurrentCalls = 4,
                PrefetchCount = 8,
                MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(5)
            });

        processor.ProcessMessageAsync += HandleAsync;
        processor.ProcessErrorAsync += args =>
        {
            logger.LogError(args.Exception, "Service Bus error from {Source} on {EntityPath}",
                args.ErrorSource, args.EntityPath);
            return Task.CompletedTask;
        };

        await processor.StartProcessingAsync(stoppingToken);
        logger.LogInformation("Processing bookstore-orders/billing");

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

        // yeni mesaj almayı bırak, elindekilerin bitmesini bekle
        await processor.StopProcessingAsync(CancellationToken.None);
    }

    private async Task HandleAsync(ProcessMessageEventArgs args)
    {
        ServiceBusReceivedMessage message = args.Message;
        OrderPlaced? order;
        try
        {
            order = message.Body.ToObjectFromJson<OrderPlaced>(MessageJson.Options);
        }
        catch (JsonException ex)
        {
            // bozuk mesajı beklemeden DLQ'ya taşı, nedenini yanına yaz
            await args.DeadLetterMessageAsync(message, "MalformedJson", ex.Message, args.CancellationToken);
            return;
        }

        await invoices.CreateInvoiceAsync(order!, args.CancellationToken);
        await args.CompleteMessageAsync(message, args.CancellationToken);
        logger.LogInformation("Completed {MessageId} (seq {Seq}, delivery {Count})",
            message.MessageId, message.SequenceNumber, message.DeliveryCount);
    }
}

Processor, receive döngüsünü, eşzamanlılığı, lock yenilemeyi ve geçici hatalarda yeniden bağlanmayı senin yerine yapar. Seçeneklerin varsayılanları, gerçek koşudan:

ReceiveMode=PeekLock AutoCompleteMessages=True MaxConcurrentCalls=1 PrefetchCount=0 MaxAutoLockRenewalDuration=00:05:00
Session: MaxConcurrentSessions=8 MaxConcurrentCallsPerSession=1

AutoCompleteMessages'ı kapatıp mesajı handler'ın sonunda elle complete ediyoruz; böylece "fatura kesildi" ile "mesaj silindi" arasındaki ilişki kodda görünür. MaxConcurrentCalls = 4 aynı anda dört handler çalıştırır, PrefetchCount sonraki mesajları önceden çeker. Prefetch'te dikkat: önceden çekilen mesajın lock'u gelir gelmez işlemeye başlar ve processor, işlenmeyi beklerken prefetch tamponundaki mesajların lock'unu yenilemez. Prefetch değerini MaxConcurrentCalls'un birkaç katından büyük tutma. Dört siparişlik gerçek log:

info: Bookstore.Billing.ServiceBus.OrderPlacedProcessor[0]
      Completed d8bdd607-dce4-473f-a78f-6bd3da0df5f7 (seq 9, delivery 1)
info: Bookstore.Billing.ServiceBus.OrderPlacedProcessor[0]
      Completed a3b78dac-7f2f-415f-8b14-145da9b3843b (seq 10, delivery 1)
info: Bookstore.Billing.ServiceBus.OrderPlacedProcessor[0]
      Completed 3622a890-9b28-41c3-8bd5-6fb37eb825c2 (seq 11, delivery 1)
info: Bookstore.Billing.ServiceBus.OrderPlacedProcessor[0]
      Completed 402ea0ce-5699-4705-b61f-6285d123caa5 (seq 12, delivery 1)

Bozuk JSON (JavaScript Object Notation) içeren bir mesajda handler mesajı hemen dead-letter'a taşıdı; DLQ'dan (Dead Letter Queue) okuduğumuzda neden bilgisi yanında:

DLQ: MessageId=bad-json-1 Reason=MalformedJson Description=Expected depth to be zero at the end of the JSON payload. There is an open JSON object or array that should be closed. Path: $.orderId | LineNumber: 0 | BytePositionInLine: 12.

PeekLock, Lock Süresi ve Lock Renewal

PeekLock modunda mesaj alınınca silinmez, belirli bir süre için kilitlenir ve diğer consumer'lara görünmez. Bu süre içinde dört cevaptan biri verilir: Complete (sil), Abandon (kilidi bırak, tekrar teslim et), DeadLetter (DLQ'ya taşı) ya da Defer (sonra sequence number ile okunmak üzere kenara koy). Lock süresi entity ayarıdır; Azure'da varsayılan 1 dakika, en fazla 5 dakika. ReceiveAndDelete modunda ise mesaj teslim anında silinir; RabbitMQ'daki autoAck: true gibi, process çökerse kaybolur.

Yazının başındaki ilk soru. Lock 30 saniye, handler 40 saniye, lock yenileme kapalı:

== no renew
  0.1s received DeliveryCount=1 LockedUntil=02:14:49
 40.4s ServiceBusException Reason=MessageLockLost
 40.5s redelivered DeliveryCount=2
== renew
  0.2s received DeliveryCount=1 LockedUntil=02:15:30
 40.2s completed
 45.2s queue empty

Lock 30. saniyede düştü, mesaj queue'ya döndü ve DeliveryCount arttı. 40. saniyede yapılan iş boşa gitti, CompleteMessageAsync MessageLockLost ile patladı ve mesaj tekrar teslim edildi. Billing için bu, aynı siparişe iki fatura demek. Varsayılan MaxAutoLockRenewalDuration (5 dakika) ile processor lock'u arka planda yeniledi ve iş temiz bitti. Ama yenileme bir garanti değil: ağ kesintisinde yenileme isteği de başarısız olabilir. İşin süresi lock süresine yaklaşıyorsa ya işi küçült ya da handler'ı idempotent yap; ikincisini serinin idempotent consumer yazısında kuracağız.

İkinci soru: handler exception fırlatıyor.

  0.2s attempt DeliveryCount=1
  0.2s ProcessError source=ProcessMessageCallback: Invoice service unavailable
  0.2s attempt DeliveryCount=2
  0.2s ProcessError source=ProcessMessageCallback: Invoice service unavailable
  0.2s attempt DeliveryCount=3
  0.2s ProcessError source=ProcessMessageCallback: Invoice service unavailable
DLQ: MessageId=poison-1 Reason=MaxDeliveryCountExceeded Description=Message could not be consumed after 3 delivery attempts.

90 saniye değil, aynı saniye içinde. Handler exception fırlatıp mesajı settle etmediğinde processor mesajı otomatik abandon ediyor (bu AutoCompleteMessages = false iken de geçerli), mesaj anında tekrar geliyor ve üç deneme milisaniyeler içinde tükeniyor. Geçici bir hata için bu, "bekle ve tekrar dene" değil, "üç kez art arda çarp ve DLQ'ya düş" demek. Gecikmeli retry'ı (scheduled message ile yeniden gönderme ya da abandon öncesi bekleme) serinin retry yazısında ele alacağız; burada akılda kalması gereken, MaxDeliveryCount'ın tek başına bir retry stratejisi olmadığı. Azure'da varsayılanı 10.

Session'larla Sıralı İşlem

Bir siparişin durum değişiklikleri (Paid, Packed, Shipped) sırayla işlenmeli, ama farklı siparişler paralel işlenebilir. RequiresSession açık bir entity'de her mesaj bir SessionId taşır; aynı session'ın mesajları sırayla ve aynı anda tek bir alıcıya teslim edilir:

foreach ((string order, string status) in new[] {
    (a, "Paid"), (b, "Paid"), (a, "Packed"), (b, "Packed"), (a, "Shipped"), (b, "Shipped") })
{
    await sender.SendMessageAsync(new ServiceBusMessage(status) { SessionId = order });
}

await using ServiceBusSessionProcessor processor = client.CreateSessionProcessor("order-status",
    new ServiceBusSessionProcessorOptions { MaxConcurrentSessions = 2, MaxConcurrentCallsPerSession = 1 });

  0.3s session=order-A status=Paid seq=1 thread=5
  0.4s session=order-B status=Paid seq=2 thread=9
  0.5s session=order-B status=Packed seq=4 thread=9
  0.5s session=order-A status=Packed seq=3 thread=9
  0.7s session=order-B status=Shipped seq=6 thread=5
  0.8s session=order-A status=Shipped seq=5 thread=9

İki sipariş birbirine karıştı ama her siparişin kendi içindeki sırası korundu. SessionId olarak sipariş id'si doğal bir seçim. MaxConcurrentCallsPerSession'ı 1'in üstüne çıkarırsan session içi sıralama garantisini kendi elinle bozarsın. Session'lar ayrıca session state (SetSessionStateAsync) tutabilir; küçük bir iş akışının durumunu mesajların yanında saklamak için kullanışlı. Sıralamanın neden pahalı bir garanti olduğu Messaging ve Event-Driven Architecture serisinin "Message Ordering ve Partitioning" yazısında.

Scheduled Message ve Duplicate Detection

Service Bus, mesajı ileri bir tarihte görünür kılabilir. "Ödeme yapılmadıysa 24 saat sonra hatırlat" gibi işler için ayrı bir scheduler kurmana gerek kalmaz:

DateTimeOffset at = DateTimeOffset.UtcNow.AddSeconds(10);
long seq = await sender.ScheduleMessageAsync(new ServiceBusMessage("payment reminder"), at);

  0.1s scheduled seq=3 for 02:14:11
 10.1s received 'payment reminder' at 02:14:11, ScheduledEnqueueTime=02:14:11

Dönen sequence number'ı saklarsan, ödeme gelince CancelScheduledMessageAsync(seq) ile hatırlatmayı iptal edebilirsin.

Duplicate detection, entity üzerinde açıldığında belirli bir zaman penceresi içinde aynı MessageId ile gelen ikinci mesajı sessizce düşürür. orders queue'sunda bu özellik 20 saniyelik pencereyle açık:

Sent attempt 1 with MessageId 580d600b-43b1-4095-b0f4-24aefca8cf95
Sent attempt 2 with MessageId 580d600b-43b1-4095-b0f4-24aefca8cf95
Received 1: attempt 1

İkinci gönderim hata vermedi, ama queue'ya girmedi. Bu özellik producer tarafındaki "gönderdim mi, göndermedim mi?" belirsizliğini çözer: timeout sonrası aynı mesajı aynı MessageId ile güvenle tekrar gönderebilirsin. İki koşulu var. Birincisi, MessageId tekrar gönderimde aynı kalmalı; her denemede Guid.NewGuid() üretirsen özellik hiçbir şey yakalamaz. Microsoft'un önerisi id'yi iş bağlamından türetmek (sipariş numarası ve mesaj konusu gibi). İkincisi, pencere sonsuz değil: varsayılanı 10 dakika, en az 20 saniye, en fazla 7 gün, ve pencere büyüdükçe throughput düşer. Duplicate detection consumer tarafındaki tekrarları (redelivery) çözmez; o iş idempotent consumer'ın.

Üretimde Dikkat

Lock süresini işin p99 süresine göre seç, ortalamasına göre değil. Uzun işlerde lock renewal'a güven ama handler'ı redelivery'ye dayanıklı yaz.

DLQ'yu izle. Dead-letter'a düşen mesaj kendiliğinden geri gelmez ve kimse bakmazsa sessizce birikir. Azure Monitor'de DLQ mesaj sayısı için alarm kur. DLQ'yu inceleme ve yeniden işleme serinin dead-letter yazısının konusu.

Tier'ı bil. Topic, subscription ve duplicate detection Basic tier'da yok; Standard ve Premium'da var. Mesaj boyutu Standard'da 256 KB, Premium'da daha büyük mesajlar destekleniyor. Mesajın içine büyük bir PDF (Portable Document Format) koyma; dosyayı blob storage'a yaz ve mesajda referansını taşı.

Processor'ların toplam eşzamanlılığını hesapla. Beş instance × MaxConcurrentCalls = 4, aynı anda yirmi fatura demek; faturanın yazıldığı veritabanı bunu kaldırabilmeli.

Bağlantı engelleri için transport'u bil. Kurumsal ağlarda 5671 portu kapalıysa ServiceBusClientOptions.TransportType = AmqpWebSockets 443 üzerinden bağlanır. Emulator bunu desteklemiyor; bu ayarı ancak Azure'da deneyebilirsin.

Sık Düşülen Tuzaklar

Her istekte yeni ServiceBusClient oluşturmak. Her client yeni bir AMQP bağlantısı açar; hem yavaştır hem de bağlantı kotasını tüketir. Client singleton, sender ve processor'lar da cache'lenir.

AutoCompleteMessages = false deyip complete etmeyi unutmak. Mesaj lock süresi dolunca tekrar gelir, MaxDeliveryCount'a ulaşınca da hiçbir hata vermeden DLQ'ya düşer. Loglarda "işlendi" görürsün, ama mesaj DLQ'dadır.

İşin lock süresini aşması. MessageLockLost sonrası iş tekrar yapılır; idempotent olmayan bir handler'da bu, çift fatura demek.

Exception fırlatmayı retry sanmak. Gördüğümüz gibi üç deneme aynı saniyede tükenir.

Filter'da body'ye bakılabileceğini sanmak. Filter'a girecek her bilgiyi ApplicationProperties'e koy.

Duplicate detection'ı her denemede yeni MessageId ile kullanmak. Özellik açık görünür, ama hiçbir şeyi yakalamaz.

Küçük Bir Deneme

★☆☆ Billing subscription'ına bir SQL filter ekle: yalnızca Currency property'si 'TRY' olanlar. Emulator'de bunu Config.json'a yazıp container'ı yeniden başlatman gerektiğini unutma.

★★☆ Handler'da exception fırlatmak yerine, hatayı yakalayıp mesajın bir kopyasını ScheduleMessageAsync ile 10 saniye sonrasına planla ve orijinalini complete et. Deneme sayısını bir ApplicationProperties alanında taşı ve üçüncü denemeden sonra mesajı elle dead-letter'a taşı.

★★★ order-status session'ında session state ile bir sipariş durum makinesi kur: Shipped, Paid'den önce gelirse mesajı defer et ve Paid gelince sequence number ile geri al.

RabbitMQ ve Azure Service Bus'ta mesaj işlenince silinir. Bir sonraki yazıda bu varsayımı bırakıyoruz: Confluent.Kafka ile .NET'te Kafka, key ile partition seçimi, consumer group'lar, offset commit ve rebalance; yani mesajın değil, okuma konumunun yönetildiği dünya.