.NET Eşzamanlılık & Paralel Programlama 36 okunma

Eşzamanlılık: Channel, Semaphore ve Concurrent

Çok iş parçacıklı uygulamalarda veri bütünlüğünü korumak için Channel<T> (producer-consumer), SemaphoreSlim (kaynak sınırlama) ve ConcurrentDictionary gibi koleksiyonların doğru kullanımı ele alınır.

Eşzamanlılık: Channel, Semaphore ve Concurrent

Eşzamanlılık: Channel, SemaphoreSlim ve Concurrent Collections ile Doğru Yol

Çok iş parçacıklı (multithread) uygulamalarda en büyük düşman lock(this) ve Monitor.Enter ile yapılan geleneksel senkronizasyondur. Bu yöntemler; deadlock'a, thread havuzu tükenmesine (starvation) ve async/await ile uyumsuzluğa yol açar. Modern .NET, bu sorunları async-first, non-blocking (engellemesiz) araçlarla çözer. İşte üç devasa kahraman:

1. Channel<T>: Producer-Consumer Deseninin Kralı

Channel<T>, adeta async/await ile çalışan güvenli bir kuyruk (Queue) gibidir. Bir veya birden fazla üretici (producer) veri yazar, bir veya birden fazla tüketici (consumer) bu veriyi bloklamadan okur.

  • Unbounded (Sınırsız) vs Bounded (Sınırlı):

    • Unbounded: Bellek patlayana kadar ekler. Üretici tüketiciden hızlıysa OutOfMemoryException alırsınız. Sadece kontrollü test ortamlarında kullanın.

    • Bounded: Kuyruk dolduğunda üreticiyi WaitToWriteAsync ile otomatik olarak bekletir (Backpressure uygular). Üretimde (prod) kesinlikle Bounded kullanın.

  • Kullanım Senaryosu: 10.000 log kaydını anında veritabanına yazmak yerine, bir Channel'a koyup arka planda 5 tüketici thread'in toplu halde (batch) yazmasını sağlayabilirsiniz.

csharp

// Kanalı oluştur (Max 100 öğe, dolarsa üreticiyi bekle)
var channel = Channel.CreateBounded<string>(
    new BoundedChannelOptions(100) { FullMode = BoundedChannelFullMode.Wait }
);
var writer = channel.Writer;
var reader = channel.Reader;

// ---- ÜRETİCİ (Producer) ----
async Task ProduceAsync(CancellationToken ct)
{
    for (int i = 0; i < 1000; i++)
    {
        await writer.WriteAsync($"Veri-{i}", ct); // Kanal dolduysa burada bekler!
        await Task.Delay(10, ct);
    }
    writer.Complete(); // Bittiğini bildir
}

// ---- TÜKETİCİ (Consumer) ----
async Task ConsumeAsync(CancellationToken ct)
{
    // ReadAllAsync, kanal tamamlanana kadar tüm verileri akış (stream) olarak getirir
    await foreach (var item in reader.ReadAllAsync(ct))
    {
        await ProcessItemAsync(item); // Veriyi işle
    }
}

// Başlatma (Çoklu tüketici desteği!)
var tasks = new List<Task>
{
    ProduceAsync(ct),
    ProduceAsync(ct), // İkinci üretici
    ConsumeAsync(ct),
    ConsumeAsync(ct), // İkinci tüketici
    ConsumeAsync(ct)  // Üçüncü tüketici
};
await Task.WhenAll(tasks);

⚠️ Kritik Uyarı: Channel içindeki öğeler referans tipiyse, tüketici okuduğunda üretici hala o nesneyi değiştiriyorsa veri bozulması olur. Immutability (Record kullanımı) veya derin kopya göndermek şarttır.

2. SemaphoreSlim: Async Uyumlu Kaynak Sınırlayıcı

Klasik lock ifadesi, bir thread'in sahibi olduğu kilitleri await'den önce bırakmasını gerektirir ve await içinde lock kullanmak derleme hatasıdır. İşte bu yüzden SemaphoreSlim hayat kurtarır. Belirli bir kaynağın (ör. API bağlantı havuzu) eşzamanlı erişim sayısını sınırlamak için birebirdir.

csharp

// Maksimum 5 eşzamanlı işlem, başlangıçta 5 boş kontenjan
private static readonly SemaphoreSlim _semaphore = new SemaphoreSlim(5, 10);

async Task CallExternalApiAsync(CancellationToken ct)
{
    // Kaynak almak için ASENKRON bekle (Thread'i bloklamaz!)
    await _semaphore.WaitAsync(ct); 
    try
    {
        // Bu blokta aynı anda en fazla 5 thread çalışabilir.
        await HttpClient.GetStringAsync("https://api.example.com", ct);
    }
    finally
    {
        // Kaynağı serbest bırak - finally içinde olması HAYATİ!
        _semaphore.Release();
    }
}

Püf Noktası: SemaphoreSlim'i using bloğuyla sarmayı unutmayın, çünkü Dispose edilmezse handle sızıntısı olur. Ayrıca Release() çağrılmadan önce exception fırlarsa kilit açılmaz; bu yüzden mutlaka try-finally kullanın (veya C# 8+ await using pattern'leri).

3. Concurrent Collections: Sihirli Değil, Dikkatli Kullanım Lazım

ConcurrentDictionary<TKey, TValue>, ConcurrentQueue<T> ve ConcurrentBag<T> kullanımı kolaylaştırır ama "Concurrent" kelimesi her derde deva değildir.

  • Doğru Kullanım: TryAdd, TryRemove, TryUpdate gibi atomik (tek hamlede) metotları kullanın.

  • Yanlış Kullanım (Feci performans ve race condition):

    csharp

    // KÖTÜ: Bu iki satır arasında başka bir thread değişiklik yapabilir!
if (!dict.ContainsKey("id"))
    dict.TryAdd("id", 1);

  • Doğru Kullanım (Atomic):

    csharp

    // GetOrAdd - 1. argüman key, 2. argüman factory metodudur.
// Ancak DİKKAT: Factory metodu birden fazla kez çağrılabilir (race condition durumunda)!
var value = dict.GetOrAdd("id", k => 0); 

// En güvenli atomik güncelleme AddOrUpdate ile:
dict.AddOrUpdate("counter",          // Key
                 1,                  // Eğer yoksa eklenecek değer
                 (key, oldValue) => oldValue + 1); // Varsa güncelleme fonksiyonu

  • ConcurrentQueue (Kuyruk) vs ConcurrentBag (Çanta): ConcurrentQueue FCFS (First Come First Serve) mantığıyla çalışır ve sipariş önemlidir. ConcurrentBag ise her thread'in kendi yerel listesine (thread-local) ekleme yapar; bu nedenle yüksek hızlı, sırasız işlemler için idealdir (ör. paralel hesaplama sonuçlarını toplama).

⚠️ Tuzak - Koleksiyonu Dolaşmak (Iteration):
ConcurrentDictionary üzerinde foreach dönerken koleksiyon değişebilir. Bu, anlık bir anlık görüntü (snapshot) alır ve sizin değişiklikleri görmemenize neden olabilir. Eğer canlı veri akışına ihtiyacınız varsa, döngü içinde TryGetValue kullanmaya devam edin veya bir ImmutableDictionary'e geçin.

4. Üçünü Birleştiren Mimariler (Pipeline Deseni)

Gerçek hayatta bu üç araç genelde birlikte kullanılır:

  1. ConcurrentQueue (veya doğrudan Channel) ile ham veri akışı sağlanır.

  2. SemaphoreSlim ile dış servis çağrıları (rate limiting) kontrol edilir.

  3. ConcurrentDictionary ile ara sonuçlar (cache) güvenle saklanır.

Sonuç ve Altın Kural:

  • lock'u unutun: Sadece çok basit, async olmayan ve milisaniye altı işlemlerde kullanın.

  • Async senkronizasyonda şampiyon: SemaphoreSlim (veya AsyncLock implementasyonları).

  • Veri akışı (pipeline): Her zaman System.Threading.Channels.

  • Paylaşılan hafıza: ConcurrentDictionary'nin atomik metotlarını (AddOrUpdate) ezberleyin.

Bu araçları doğru kullandığınızda, uygulamanız 100 thread tarafından aynı anda saldırıya uğrasa bile veri bütünlüğü bozulmaz ve kilitlenme (deadlock) yaşamazsınız.

Tüm yazılar