Asenkron İşleme: Queue, Worker, Fan-Out ve Backpressure

Sürüm notu: Bu yazıdaki kodlar C# 14 ve .NET 10 ile derlenip denenmiştir; örnek ASP.NET Core 10 minimal API (Application Programming Interface), System.Threading.Channels ve BackgroundService ile koşturuldu.

Önce Tahmin Et

Bookstore'da kurumsal müşteriler ay sonunda faturalarını PDF olarak indiriyor. Endpoint basit: POST /orders/{id}/invoice PDF'i üretiyor, dosyanın adresini dönüyor. Üretim CPU'ya (Central Processing Unit) bağlı, bir fatura yaklaşık 400 ms sürüyor ve sunucu aynı anda en fazla dört tane üretebiliyor. Ay sonunun ilk dakikasında 100 istek aynı anda geliyor; istemci tarafındaki HttpClient 5 saniye bekleyip vazgeçiyor.

Kaç müşteri faturasını alır? Sunucu kaç PDF üretir? Ve endpoint'i "işi queue'ya bırak, hemen cevap dön" şekline çevirirsek, faturalar daha hızlı mı üretilir?

Request Yolundan Ağır İşi Çıkarmak

Bir HTTP (Hypertext Transfer Protocol) isteğinin içinde yapılan her iş, istemcinin beklediği süreye ve sunucunun o sırada tuttuğu kaynaklara eklenir. Bazı işler cevabın parçası değildir: fatura PDF'i, sipariş onayı e-postası, arama index'inin güncellenmesi, raporun hazırlanması. Bunların ortak özellikleri yavaş olmaları, çoğu zaman dış bir sisteme bağlı olmaları ve kullanıcının sonucu birkaç saniye sonra görmesinin sorun olmaması.

Bu işleri request yolundan çıkarmanın standart şekli queue + worker: API işi tarif eden küçük bir mesajı queue'ya bırakır ve hemen cevap verir, ayrı çalışan worker'lar mesajları kendi hızlarında işler. Microsoft Learn'deki Cloud Design Patterns kataloğu bunun iki yüzünü ayrı pattern'ler olarak anlatıyor. Queue-Based Load Leveling, ani yükü queue'da biriktirip worker'ların sabit hızda işlemesi; Asynchronous Request-Reply de istemcinin işin sonucunu sonradan nasıl öğreneceği. Bookstore'un mimari kararlarında sipariş e-postası zaten bir in-process Channel<T> üzerinden gidiyordu; bu yazıda aynı fikri fatura için genişletiyoruz.

Deneme: Senkron Fatura

Önce bugünkü hâl. PDF üretimini, aynı anda dört işe izin veren bir SemaphoreSlim ve 400 ms'lik bir bekleme ile taklit ediyoruz:

// PDF üretimi CPU'ya bağlı: aynı anda en fazla 4 tane, her biri ~400 ms
public sealed class InvoiceRenderer
{
    private readonly SemaphoreSlim _cores = new(4);
    private readonly ConcurrentDictionary<int, bool> _invoices = new();
    private int _rendered, _skipped;

    public async Task RenderAsync(int orderId)
    {
        if (_invoices.ContainsKey(orderId)) { Interlocked.Increment(ref _skipped); return; }   // fatura zaten var
        await _cores.WaitAsync();
        try
        {
            await Task.Delay(400);
            _invoices[orderId] = true;
            Interlocked.Increment(ref _rendered);
        }
        finally { _cores.Release(); }
    }
}

// eski yol: PDF request'in içinde üretiliyor
app.MapPost("/orders/{orderId:int}/invoice/sync", async (int orderId, InvoiceRenderer renderer) =>
{
    await renderer.RenderAsync(orderId);
    return Results.Ok($"/invoices/{orderId}.pdf");
});

100 eşzamanlı istek, 5 saniyelik client timeout:

sync : 200 OK 48, timed out 52, p50 2,856 ms, slowest OK 4,866 ms
       server rendered 100 PDFs, 52 for clients that had already given up

Saniyede 10 fatura üretilebiliyor (4 ÷ 0,4 s), yani 100 faturanın sonuncusu 10. saniyede hazır. İstemciler 5. saniyede vazgeçtiği için yalnız 48'i cevap aldı; başarılı isteklerin ortancası bile 2,9 saniye. Daha kötüsü ikinci satır: sunucu 100 PDF'in hepsini üretti, 52'si artık bekleyeni olmayan istemciler için. İstemci ne yapar? Tekrar dener. Bir sonraki dakikada aynı 52 istek, öncekilerin bitmemiş işinin üstüne yeniden gelir; senkron tasarım yük altında kendi yükünü büyütüyor. (Request'in CancellationToken'ını üretime geçirmek boşa giden işi azaltır ama bekleme süresini değiştirmez.)

Queue, Worker ve 202 Accepted

Yeni tasarımda endpoint işi bir bounded Channel<T>'a yazıyor ve 202 Accepted ile bir status adresi dönüyor. Dört worker queue'yu tüketiyor:

// bounded queue: en fazla 50 iş bekleyebilir, fazlası reddedilir
builder.Services.AddSingleton(Channel.CreateBounded<InvoiceJob>(new BoundedChannelOptions(50) { FullMode = BoundedChannelFullMode.Wait }));
builder.Services.AddHostedService<InvoiceWorker>();

// yeni yol: işi queue'ya bırak, 202 ve status adresi dön
app.MapPost("/orders/{orderId:int}/invoice", (int orderId, HttpRequest request, InvoiceJobs jobs, Channel<InvoiceJob> queue) =>
{
    string key = request.Headers["Idempotency-Key"].FirstOrDefault() ?? Guid.NewGuid().ToString();
    if (jobs.TryGetByKey(key, out InvoiceJob? existing))
        return Results.Accepted($"/invoice-jobs/{existing.Id}");   // aynı istek tekrar geldi: yeni iş yok

    InvoiceJob job = jobs.Create(key, orderId);
    if (!queue.Writer.TryWrite(job))
    {
        jobs.Remove(job, key);
        request.HttpContext.Response.Headers.RetryAfter = "5";
        return Results.StatusCode(StatusCodes.Status503ServiceUnavailable);   // backpressure
    }
    request.HttpContext.Response.Headers.RetryAfter = "2";
    return Results.Accepted($"/invoice-jobs/{job.Id}");
});

app.MapGet("/invoice-jobs/{id:guid}", (Guid id, InvoiceJobs jobs) =>
    jobs.Find(id) switch
    {
        null => Results.NotFound(),
        { Status: JobStatus.Done } job => new SeeOther($"/invoices/{job.OrderId}.pdf"),   // iş bitti: sonuca yönlendir
        { } job => Results.Ok(new { status = job.Status.ToString() }),
    });

public sealed class InvoiceWorker(Channel<InvoiceJob> queue, InvoiceRenderer renderer) : BackgroundService
{
    protected override Task ExecuteAsync(CancellationToken stoppingToken) =>
        Task.WhenAll(Enumerable.Range(0, 4).Select(_ => ConsumeAsync(stoppingToken)));

    private async Task ConsumeAsync(CancellationToken stoppingToken)
    {
        await foreach (InvoiceJob job in queue.Reader.ReadAllAsync(stoppingToken))
        {
            job.Status = JobStatus.Running;
            await renderer.RenderAsync(job.OrderId);
            job.Status = JobStatus.Done;
        }
    }
}

SeeOther, 303 See Other ve Location header'ı yazan küçük bir IResult. Aynı 100 istek:

async: 202 Accepted 54, 503 rejected 46, p50 8 ms, p99 9 ms
       queue drained in 5.7 s by 4 workers, 54 PDFs rendered

Kabul edilen her istek 9 ms içinde cevap aldı; istemci beklemiyor, bağlantı tutulmuyor, timeout yok. Ama açılıştaki son sorunun cevabı hayır: faturalar daha hızlı üretilmedi. 54 PDF 5,7 saniyede bitti, yine saniyede ~10 fatura; worker'lar senkron endpoint'le aynı dört "çekirdeği" kullanıyor. Asenkron işleme işi hızlandırmaz, işin zamanlamasını request'ten ayırır: kullanıcı bekletilmez, ani yük zamana yayılır ve kapasite ayrı ölçeklenebilir hâle gelir.

Status endpoint'i Microsoft Learn'deki Asynchronous Request-Reply pattern'inin önerdiği gibi davranıyor: iş sürerken 200 ve durum, bitince 303 See Other ile sonuca yönlendirme. Bir işin gerçek polling çıktısı:

poll 1: 200 {"status":"Running"}
poll 2: 200 {"status":"Running"}
poll 3: 200 {"status":"Running"}
poll 4: 303 Location: /invoices/2001.pdf

Pattern'in 302 değil 303 önermesinin sebebi, 303'ün istemciye yönlendirilen adresi her zaman GET ile istemesini söylemesi; 302'de bazı istemciler orijinal POST'u tekrarlayabiliyor. 202 cevabındaki Retry-After da istemciye ne sıklıkla sorması gerektiğini söylüyor. Polling yerine sonucu push etmek istersen (SignalR, webhook, e-posta) bu ilerideki Backend kategorisindeki Real-time Uygulamalar serisinin konusu.

Backpressure: Bounded Queue

Çıktıdaki ikinci sayı 46 adet 503. Queue 50 işle sınırlı; dört worker'ın aldığı dört iş ve queue'daki 50 iş dolunca yeni istekler TryWrite'ta reddedildi ve istemciye 503 Service Unavailable ile Retry-After: 5 döndü. Buna backpressure deniyor: tüketici yetişemediğinde üreticiye "yavaşla" sinyali göndermek.

Unbounded bir queue bu sinyali vermez, ama yükü de ortadan kaldırmaz. Sadece sorunu ileri iter: queue bellekte büyür, kabul edilen bir işin tamamlanma süresi dakikalara, saatlere çıkar ve "202 Accepted" verilen işin ne zaman biteceği bilinmez hâle gelir. Process yeniden başlarsa bellekteki bütün iş kaybolur. Bounded queue'da kapasite bir tasarım kararı: 50 iş, dört worker ile en fazla ~5 saniyelik bekleme demek. İstemciye "şimdi değil, 5 saniye sonra" demek, "kabul ettim" deyip 10 dakika sonra bitirmekten daha dürüst.

BoundedChannelOptions.FullMode queue dolduğunda ne olacağını belirliyor: Wait yazanı bekletir (bizim örnekte TryWrite beklemeden false döndüğü için reddediyoruz), DropOldest ve DropNewest queue'daki bir işi atar, DropWrite yazılmak istenen işi sessizce düşürür. Fatura gibi kaybolmaması gereken işlerde sessiz düşürme seçenekleri yanlış; telemetri ya da "son konum" gibi yalnız en yeni değerin önemli olduğu akışlarda doğru olabilir.

Queue Derinliği ve Autoscaling

Queue'nun derinliği sistemin en dürüst metriği: geliş hızı işleme hızını geçtiğinde büyür, geçmediğinde boşalır. Kaç worker gerektiğini serinin ilk yazısındaki Little's law söylüyor: gerekli eşzamanlılık = geliş hızı × işlem süresi. Ay sonunda 10 dakikada 2.000 fatura geliyorsa geliş hızı saniyede 3,3; 0,4 saniyelik işlemle 3,3 × 0,4 = 1,3 eşzamanlı işlem, yani iki worker kararlı durumda yetiyor. Aynı anda gelen 100 iş ise kapasite ne olursa olsun bir queue'da birikiyor ve boşalma süresi derinlik ÷ işleme hızı: 100 ÷ 10 = 10 saniye.

Bu hesap autoscaling kuralına dönüşebiliyor: "queue'da kalan işin boşalma süresi 30 saniyeyi geçerse worker ekle". Kubernetes'te KEDA (Kubernetes Event-driven Autoscaling) bunu RabbitMQ, Azure Service Bus ya da Kafka queue derinliğine göre pod sayısını değiştirerek yapıyor. CPU'ya göre ölçeklemek bu iş için geç kalır: worker'lar zaten doluyken CPU yüksektir, queue'da bekleyen iş sayısı ise sorunu daha önce gösterir. Bir uyarı: worker'ları ölçeklemek, worker'ların konuştuğu veritabanını ölçeklemez; serinin ilk yazısındaki connection pool hesabı worker sayısı için de geçerli.

Fan-Out: Bir Event, Birden Fazla İş

Sipariş verildiğinde yalnız fatura değil, onay e-postası, stok rezervasyonu ve analitik kaydı da gerekiyor. Bunların hepsini OrderService'ten sırayla çağırmak, her yeni iş için sipariş koduna dokunmak demek. Fan-out'ta sipariş tek bir OrderPlaced event'i yayınlar ve her ilgili iş kendi subscription'ı üzerinden kendi kopyasını alır: publish/subscribe. Queue (point-to-point) ile topic (pub/sub) farkı burada önemli: queue'daki bir mesajı worker'lardan yalnız biri alır (competing consumer), topic'teki bir mesajın kopyası her subscriber'a gider.

Fan-out'un tasarıma iki etkisi var. Her subscriber kendi hızında ve kendi hatasıyla yaşar: e-posta sağlayıcısı çöktüğünde fatura üretimi etkilenmez. Ama her subscriber bağımsız olarak geride kalabilir, o yüzden "sipariş tamamlandı mı?" sorusu artık tek bir yerde cevaplanmaz. Bu kavramların derini Messaging ve Event-Driven Architecture serisinin "Channel'lar: Point-to-Point Queue, Publish/Subscribe Topic ve Competing Consumer'lar" yazısında.

At-Least-Once ve Idempotency

Gerçek message broker'ların neredeyse hepsi at-least-once teslim eder: worker bir mesajı işleyip onayını (ack) göndermeden çökerse mesaj başka bir worker'a yeniden verilir. Yani aynı fatura işi iki kez gelebilir. Tasarımın buna dayanması gerekiyor ve iki yerde idempotency şart.

Birincisi API girişi. İstemci 202 cevabını network yüzünden alamazsa isteği tekrarlar; sunucu bunun aynı istek olduğunu bilmezse ikinci bir iş açar. Çözüm istemcinin her mantıksal işlem için bir Idempotency-Key header'ı göndermesi ve sunucunun aynı key'e aynı işi dönmesi. Bu header IETF'te (Internet Engineering Task Force) hâlâ bir draft olarak duruyor, ama Microsoft Learn'deki pattern sayfası da tam olarak bunu öneriyor:

idempotency: 202 /invoice-jobs/099377df-abcd-4bbb-a99d-f488c577c8d5 | 202 /invoice-jobs/099377df-abcd-4bbb-a99d-f488c577c8d5 | same job: True

İkincisi worker. Aynı mesaj ikinci kez geldiğinde worker işi tekrar yapmamalı ya da tekrar yapması zararsız olmalı. Örnekteki RenderAsync, faturası zaten üretilmiş bir siparişi atlıyor:

redelivery: rendered 1, skipped as duplicate 1

Gerçek bir sistemde bu kontrol bellekteki bir dictionary değil, veritabanında bir unique constraint (Invoices.OrderId) ya da işlenmiş mesajların tutulduğu bir tablo olur; kontrol ile yazmanın aynı transaction'da olması gerekir. Bu konunun tamamı Messaging ve Event-Driven Architecture serisinin "Delivery Guarantee'leri: At-Most-Once, At-Least-Once ve "Exactly-Once" Gerçeği" ve "Idempotent Consumer: Aynı Message İki Kez Gelirse Ne Olur?" yazılarında.

Sık Düşülen Tuzaklar

In-memory queue'yu kalıcı sanmak. Örnekteki Channel<T> process'in belleğinde; deploy ya da çökme anında queue'daki her iş kaybolur ve 202 verilmiş işlerin status'u sonsuza kadar "Queued" kalır. Kaybolmaması gereken işler için kalıcı bir broker (RabbitMQ, Azure Service Bus) ve mesajı veritabanı transaction'ıyla aynı anda yazan outbox gerekir; .NET client'ları ilerideki Backend kategorisindeki Message Queue'larla Çalışmak serisinde.

Unbounded queue ile backpressure'ı kaldırmak. Yük yok olmaz, queue'da birikir; sınır koy ve dolunca dürüstçe reddet.

Status endpoint'i olmadan 202 dönmek. İstemci işin bitip bitmediğini öğrenemezse 202 bir cevap değil, bir kara deliktir.

Poison message'ı sonsuza kadar denemek. Her seferinde exception fırlatan bir mesaj, worker'ı meşgul eder ve arkasındaki işleri bekletir; belirli sayıda denemeden sonra dead-letter queue'ya ayır.

Queue'nun sırayı koruduğunu varsaymak. Birden fazla worker paralel çalıştığında işler geliş sırasından farklı bir sırada biter; sıra önemliyse partition key ile aynı key'in işlerini aynı worker'a gönder.

Her şeyi asenkron yapmak. Kullanıcının sonucu hemen görmesi gereken işler (ödeme onayı, stok kontrolü) asenkron yapılınca kullanıcı deneyimi polling ve belirsizlikle dolar; asenkronluğu cevabın parçası olmayan işlere sakla.

Küçük Bir Deneme

★☆☆ Queue kapasitesini 50'den 200'e çıkar ve aynı 100 isteği gönder. Kaç istek kabul edildi, queue kaç saniyede boşaldı? Kapasiteyi büyütmenin istemcinin beklediği süreye etkisini Little's law ile açıkla.

★★☆ Worker sayısını 4'ten 8'e çıkar ama InvoiceRenderer'daki semaphore'u 4'te bırak. Boşalma süresi değişti mi? Neden? Darboğazın worker sayısı değil başka bir kaynak olduğu gerçek bir örnek düşün.

★★★ Worker'a bir hata senaryosu ekle: orderId 13'ün katı olan faturalar exception fırlatsın. Üç denemeden sonra işi bir dead-letter listesine ayıran, status endpoint'inde Failed ve RFC (Request for Comments) 9457 Problem Details formatında bir hata dönen yapıyı kur.

Bu yazıyla serinin yapı taşlarının ilk yarısı tamamlandı: estimation, scaling, load balancing, caching, veri katmanı, consistent hashing ve asenkron işleme. Bir sonraki yazıda bu yapı taşlarından birini uçtan uca tasarlıyoruz: token bucket'tan sliding window'a rate limiting algoritmaları ve ASP.NET Core rate limiter'ının instance başına sayaç tutması yüzünden neden distributed bir rate limiter gerektiği.