C# Channels ile Producer/Consumer: Backpressure, Bounded Channel ve Pipeline

Sürüm notu: Bu yazıdaki kodlar C# 10 ve .NET 6 API'leriyle yazılmıştır; .NET 10 SDK ile derlenip denenmiştir.

Önce Tahmin Et

Bir thread sipariş üretiyor, diğeri paketliyor. Aralarında en fazla iki sipariş alabilen bir channel var. Producer her yazdığı siparişi ekrana basıyor, paketleyici her siparişe 100 ms harcıyor. "Wrote 1003" satırı ekranda nerede görünür: "Processed 1001"den önce mi, sonra mı? Ya "Wrote 1005"?

using System.Threading.Channels;

Channel<int> orders = Channel.CreateBounded<int>(capacity: 2);

Task producer = Task.Run(async () =>
{
    for (int orderId = 1001; orderId <= 1005; orderId++)
    {
        await orders.Writer.WriteAsync(orderId);
        Console.WriteLine($"Wrote     {orderId}");
    }
    orders.Writer.Complete();
});

await Task.Delay(50);
await foreach (int orderId in orders.Reader.ReadAllAsync())
{
    await Task.Delay(100); // siparişi paketlemek
    Console.WriteLine($"Processed {orderId}");
}
await producer;
Console.WriteLine("All orders processed");

Channel: async Bir Queue

Beş çalıştırmada da aynı çıktıyı aldım:

Wrote     1001
Wrote     1002
Wrote     1003
Processed 1001
Wrote     1004
Processed 1002
Wrote     1005
Processed 1003
Processed 1004
Processed 1005
All orders processed

İlk üç yazma beklemeden geçti. Kapasite iki olduğu hâlde üç siparişin sığmasının sebebi, paketleyicinin 1001'i channel'dan hemen alıp 100 ms'lik işine başlaması: 1001 artık channel'da değil, paketleyicinin elinde. 1002 ve 1003 channel'ı doldurdu, 1004'ün WriteAsync'i ise beklemeye geçti. Ondan sonra çıktı bir ritme oturuyor: her "Processed" bir yer açıyor, hemen ardından bir "Wrote" geliyor. Producer, consumer'ın hızına kendiliğinden ayak uydurdu. Buna backpressure denir ve bu yazının en önemli kavramı.

System.Threading.Channels, .NET Core 3.0'dan beri framework'ün içinde. Bir channel'ı thread-safe ve async bir queue gibi düşün: bir ucunda yazanlar (ChannelWriter<T>), diğer ucunda okuyanlar (ChannelReader<T>) var. Önceki yazıdaki ConcurrentQueue<T>'den farkı, queue boşken okuyanın, doluyken yazanın thread bloklamadan await ile beklemesi. BlockingCollection<T> aynı işi yapar ama bekleyen thread'i bloklar; async koda uygun değildir.

Writer ve Reader

En basit hâli sınırsız (unbounded) bir channel. E-posta gönderim queue'su düşün:

using System.Threading.Channels;

Channel<string> emails = Channel.CreateUnbounded<string>();

// yazan taraf: açık bir unbounded channel'da TryWrite hiç başarısız olmaz
emails.Writer.TryWrite("order-confirmation:1001");
emails.Writer.TryWrite("order-confirmation:1002");
emails.Writer.TryWrite("shipping-notice:1001");
emails.Writer.Complete();

Console.WriteLine($"Waiting in channel: {emails.Reader.Count}");

// okuyan taraf: ReadAllAsync'in sarmaladığı alt seviye döngü
while (await emails.Reader.WaitToReadAsync())
{
    while (emails.Reader.TryRead(out string? email))
        Console.WriteLine($"Sending {email}");
}

await emails.Reader.Completion;
Console.WriteLine($"Completed: {emails.Reader.Completion.IsCompleted}, can write: {emails.Writer.TryWrite("late")}");

Waiting in channel: 3
Sending order-confirmation:1001
Sending order-confirmation:1002
Sending shipping-notice:1001
Completed: True, can write: False

Yazan taraf iki şey söyler: "al bunu" (TryWrite ya da WriteAsync) ve "bitti, başka bir şey gelmeyecek" (Complete). İkincisi en az birincisi kadar önemli; okuyanın döngüden ne zaman çıkacağını o belirler. Okuyan tarafta WaitToReadAsync, okunacak bir şey gelene kadar ya da channel tamamlanana kadar bekler; false dönerse hem queue boştur hem de yazan taraf işini bitirmiştir. İçteki TryRead döngüsü, bir uyanışta queue'da biriken her şeyi tek seferde çeker. ReadAllAsync() tam olarak bu döngüyü bir IAsyncEnumerable<T> olarak sarar; serinin önceki yazısında gördüğümüz await foreach ile okunur. Günlük kodda ReadAllAsync yeterli; her bir okumada kendi kontrolünü istediğinde alt seviye döngüye inersin. Reader.Completion ise channel tamamen boşalıp kapandığında biten bir task.

Bounded Channel ve Backpressure

Unbounded channel'ın bir riski var: producer consumer'dan hızlıysa queue sınırsız büyür ve bunu sana bir exception değil, artan bellek kullanımı söyler. Bounded channel'da queue dolduğunda ne olacağını FullMode belirler. Stok güncellemeleri akan, kapasitesi üç olan bir channel'a beş güncellemeyi TryWrite ile yazalım:

using System.Threading.Channels;

foreach (BoundedChannelFullMode mode in new[] { BoundedChannelFullMode.DropOldest, BoundedChannelFullMode.DropNewest, BoundedChannelFullMode.DropWrite, BoundedChannelFullMode.Wait })
{
    var dropped = new List<int>();
    Channel<int> stockUpdates = Channel.CreateBounded<int>(
        new BoundedChannelOptions(3) { FullMode = mode },
        item => dropped.Add(item));

    var accepted = new List<bool>();
    for (int update = 1; update <= 5; update++)
        accepted.Add(stockUpdates.Writer.TryWrite(update));
    stockUpdates.Writer.Complete();

    var kept = new List<int>();
    await foreach (int update in stockUpdates.Reader.ReadAllAsync())
        kept.Add(update);

    Console.WriteLine($"{mode,-10} kept [{string.Join(", ", kept)}]  dropped [{string.Join(", ", dropped)}]  TryWrite: {string.Join(" ", accepted)}");
}

DropOldest kept [3, 4, 5]  dropped [1, 2]  TryWrite: True True True True True
DropNewest kept [1, 2, 5]  dropped [3, 4]  TryWrite: True True True True True
DropWrite  kept [1, 2, 3]  dropped [4, 5]  TryWrite: True True True True True
Wait       kept [1, 2, 3]  dropped []  TryWrite: True True True False False

Varsayılan mod Wait: queue doluyken TryWrite false döner, WriteAsync yer açılana kadar bekler. Tahmin sorusundaki ritim buydu. Diğer üç mod beklemek yerine bir şeyi atar: DropOldest en eski element'i, DropNewest queue'daki en yeni element'i, DropWrite ise yazılmaya çalışılanı. Hangi element'in atıldığını itemDropped callback'iyle öğrenebilirsin; bu overload .NET 6 ile geldi. Seçim işin doğasına bağlı. Sipariş kaybedilemez, Wait kullanırsın. Sadece son değerin önemli olduğu bir stok ya da sıcaklık stream'inde DropOldest mantıklıdır: queue dolmuşsa eski güncellemeler zaten bayattır.

Birden Çok Consumer

Paketleme yavaşsa paketleyici sayısını artırırsın. Aynı reader'dan birden çok consumer okuyabilir; her element yalnızca birine gider:

using System.Collections.Concurrent;
using System.Threading.Channels;

Channel<int> orders = Channel.CreateBounded<int>(new BoundedChannelOptions(10)
{
    SingleWriter = true,
    SingleReader = false,
});

var processedBy = new ConcurrentDictionary<string, int>();

Task[] packers = Enumerable.Range(1, 3)
    .Select(packerNo => Task.Run(async () =>
    {
        await foreach (int orderId in orders.Reader.ReadAllAsync())
        {
            await Task.Delay(orderId % 3 == 0 ? 30 : 10); // bazı siparişlerin paketlenmesi daha uzun sürüyor
            processedBy.AddOrUpdate($"packer-{packerNo}", 1, (_, count) => count + 1);
        }
    }))
    .ToArray();

for (int orderId = 1; orderId <= 60; orderId++)
    await orders.Writer.WriteAsync(orderId);
orders.Writer.Complete();

await Task.WhenAll(packers);

foreach (KeyValuePair<string, int> pair in processedBy.OrderBy(p => p.Key))
    Console.WriteLine($"{pair.Key}: {pair.Value} orders");
Console.WriteLine($"Total: {processedBy.Values.Sum()}");

packer-1: 21 orders
packer-2: 17 orders
packer-3: 22 orders
Total: 60

İş, kimin boş olduğuna göre kendiliğinden dağıldı; uzun süren siparişlere denk gelen paketleyici daha az iş aldı. Elle bir dağıtım mantığı yazmadık. SingleWriter ve SingleReader seçenekleri channel'a verdiğin bir sözdür: "bu channel'a tek bir yerden yazılacak", "tek bir yerden okunacak". Implementasyon bu sözlere güvenip daha hızlı bir yol seçebilir; sözü tutmazsan davranış tanımsızdır. Emin değilsen varsayılan false değerlerinde bırak.

Bir bedel var: birden çok consumer varken işlerin bitiş sırası garanti değildir. 1002 numaralı sipariş 1001'den önce paketlenebilir. Sıra önemliyse tek consumer kullanırsın ya da sıralamayı bir sonraki aşamada yeniden kurarsın.

Pipeline Kurmak ve Hatayı İletmek

Channel'lar uç uca eklendiğinde bir pipeline olur. Siparişler alınır, stok ayrılır, fatura kesilir; her aşama kendi hızında çalışır ve aralarındaki bounded channel'lar hızlı aşamanın yavaş olanı ezmesini engeller:

using System.Globalization;
using System.Threading.Channels;

CultureInfo.CurrentCulture = CultureInfo.InvariantCulture;

var stock = new Dictionary<string, int> { ["978-0132350884"] = 2, ["978-0135957059"] = 5 };
Order[] incoming =
{
    new(1001, "978-0132350884", 1, 37.90m),
    new(1002, "978-0135957059", 2, 42.50m),
    new(1003, "978-0132350884", 2, 37.90m),
    new(1004, "978-0000000000", 1, 10.00m),
};

Channel<Order> placed = Channel.CreateBounded<Order>(5);
Channel<Order> reserved = Channel.CreateBounded<Order>(5);

Task intake = Task.Run(async () =>
{
    foreach (Order order in incoming)
        await placed.Writer.WriteAsync(order);
    placed.Writer.Complete();
});

Task reservation = Task.Run(async () =>
{
    try
    {
        await foreach (Order order in placed.Reader.ReadAllAsync())
        {
            if (!stock.TryGetValue(order.Isbn, out int available))
                throw new InvalidOperationException($"Unknown ISBN {order.Isbn} in order {order.Id}");
            if (available < order.Quantity)
            {
                Console.WriteLine($"Order {order.Id}: out of stock");
                continue;
            }
            stock[order.Isbn] = available - order.Quantity;
            await reserved.Writer.WriteAsync(order);
        }
        reserved.Writer.Complete();
    }
    catch (Exception ex)
    {
        reserved.Writer.Complete(ex); // hatayı bir sonraki aşamaya ilet
    }
});

try
{
    await foreach (Order order in reserved.Reader.ReadAllAsync())
        Console.WriteLine($"Invoice for order {order.Id}: {order.Quantity * order.UnitPrice:F2}");
}
catch (InvalidOperationException ex)
{
    Console.WriteLine($"Pipeline stopped: {ex.Message}");
}

await Task.WhenAll(intake, reservation);

record Order(int Id, string Isbn, int Quantity, decimal UnitPrice);

Order 1003: out of stock
Invoice for order 1001: 37.90
Invoice for order 1002: 85.00
Pipeline stopped: Unknown ISBN 978-0000000000 in order 1004

Üç noktaya dikkat et. Birincisi, stock dictionary'sine yalnızca rezervasyon aşaması dokunuyor; tek bir task'ın sahip olduğu state için lock'a gerek yok. Pipeline'ın güzel taraflarından biri bu: shared state yerine her aşamanın kendi state'i var, aşamalar arasında sadece mesajlar akıyor. İkincisi, her aşama işi bitince kendi çıkış channel'ını tamamlıyor; tamamlanma, su gibi pipeline boyunca akıyor. Üçüncüsü hata. Bilinmeyen ISBN rezervasyon aşamasını durdurdu; aşama exception'ı yutmak yerine Complete(ex) ile bir sonraki aşamaya iletti. Fatura aşaması önce queue'da bekleyen 1001 ve 1002'yi işledi, sonra ReadAllAsync aynı exception'ı fırlattı. Complete'i bir aşamanın her çıkış yolunda çağırmazsan aşağıdaki aşamalar sonsuza kadar bekler.

Channel mı, Başka Bir Şey mi

Elinde hazır bir liste varsa ve her element'i sınırlı concurrency ile işlemek istiyorsan channel'a gerek yok; Parallel.ForEachAsync daha kısa. Channel, üretimin ve tüketimin zamana yayıldığı, farklı hızlarda aktığı yerde parlar: arka planda çalışan bir e-posta gönderici, gelen webhook'ları sıraya alan bir worker, log ya da metrik toplayıcı. ASP.NET Core'da bunun tipik hâli, controller'ın channel'a yazdığı ve arka planda çalışan bir BackgroundService'in okuduğu bir queue. Bir şeyi bilerek seç: channel bellekte yaşar. Uygulama çökerse ya da yeniden başlarsa içindekiler gider. Kaybedilmemesi gereken işler için RabbitMQ, Azure Service Bus gibi kalıcı bir queue ya da veritabanı tablosu gerekir; channel onların yerini tutmaz.

Sık Düşülen Tuzaklar

Complete'i unutmak. Program hata vermez, sadece bitmez:

using System.Threading.Channels;

Channel<int> orders = Channel.CreateUnbounded<int>();
orders.Writer.TryWrite(1001);
orders.Writer.TryWrite(1002);
// unutuldu: orders.Writer.Complete();

Task consumer = Task.Run(async () =>
{
    await foreach (int orderId in orders.Reader.ReadAllAsync())
        Console.WriteLine($"Processed {orderId}");
    Console.WriteLine("Consumer finished");
});

try
{
    await consumer.WaitAsync(TimeSpan.FromSeconds(1));
}
catch (TimeoutException)
{
    Console.WriteLine("Consumer is still waiting after 1 second");
}

orders.Writer.Complete();
await consumer;

Console.WriteLine($"TryComplete again: {orders.Writer.TryComplete()}");
try
{
    orders.Writer.Complete();
}
catch (ChannelClosedException ex)
{
    Console.WriteLine($"Complete again: {ex.GetType().Name}: {ex.Message}");
}

Processed 1001
Processed 1002
Consumer is still waiting after 1 second
Consumer finished
TryComplete again: False
Complete again: ChannelClosedException: The channel has been closed.

Consumer iki siparişi işledi ve sonra "belki bir şey daha gelir" diye beklemeye devam etti. Bunu fark etmenin tek yolu çoğu zaman uygulamanın kapanmaması oluyor. Producer'da Complete'i finally bloğunda çağırmak iyi bir alışkanlık. Çıktının sonundaki iki satır da ayrı bir tuzak: ikinci kez Complete çağırmak ChannelClosedException atar. Birden çok yerden kapatılabilecek bir channel'da TryComplete kullan. Kapanmış bir channel'a WriteAsync ile yazmak da aynı exception'ı fırlatır.

DropWrite ve DropNewest'te TryWrite'ın sonucuna güvenmek. Yukarıdaki çıktıya bak: element atılsa bile TryWrite true döndü. Kayıptan haberdar olmak istiyorsan itemDropped callback'ini kullan.

Her yerde unbounded channel kullanmak. Kolay ama queue'nun büyümesini kimse durdurmaz. Producer senin kontrol etmediğin bir kaynaksa (HTTP istekleri, dış sistemden gelen event'ler) bounded channel ve bilinçli bir FullMode seç.

Consumer'da exception'ı yakalamamak. Task.Run içindeki bir consumer exception atarsa task faulted olur, okuma durur ve kimse await etmiyorsa hata sessizce kaybolur; producer da queue dolunca sonsuza kadar bekler. Consumer task'larını mutlaka bir yerde await et ve tek bir element'teki hatanın bütün consumer'ı öldürüp öldürmeyeceğine bilerek karar ver.

Küçük Bir Deneme

★☆☆ Tahmin sorusundaki kapasiteyi 1'e, sonra 10'a çıkar. "Wrote" satırlarının yeri nasıl değişiyor? Toplam süre değişiyor mu?

★★☆ p14d'deki paketleyicilere bir hata ekle: 37 numaralı sipariş exception atsın. Sadece o siparişi loglayıp devam eden ve hatayı bütün pipeline'a yayan iki ayrı sürüm yaz. Her ikisinde de program sonunda kaç sipariş işlenmiş oluyor?

★★★ p14e'deki pipeline'a dördüncü bir aşama ekle: kesilen faturaları beşerli gruplar hâlinde "toplu e-posta" olarak gönder. Son grup beşten küçükse ne yapmalı? İpucu: WaitToReadAsync ile TryRead döngüsünü kullan.

Bu yazıda bir şeyi hep Task.Delay ile taklit ettik: "100 ms iş yapıyormuş gibi". Bir sonraki yazıda bekleme ve zamanlamanın kendisine bakıyoruz: Task.Delay'in nasıl çalıştığı, .NET'in üç farklı timer'ı arasındaki farklar, .NET 6 ile gelen PeriodicTimer ve herhangi bir task'a timeout ekleyen WaitAsync.