Channels ve concurrent

C# kursu · Uzman · Ders 45/47

Channels ve concurrent koleksiyonlarla paralel akış yazarsın.

Channels ve concurrent

Bu derste logları bir Channel akışından geçirip ConcurrentDictionary'de sayacağız. Üretici her logu WriteAsync ile yazar, Complete ile kapatır; tüketici ReadAllAsync ile her logu okur ve AddOrUpdate ile sayar.

Bu, önceki derslerde öğrendiğin ParseWithSpan'ı da kullanır: her logun seviyesi kopyasız çıkarılır. Sonuçlar alfabetik sıralanır ve deterministik çıkar.

Run metodu tasarımı

Run metodu async Task olarak tanımlanır. İçinde bir sınırsız kanal ve bir ConcurrentDictionary oluşturulur. Loglar tek tek yazılır, kanal tamamlanır, sonra await foreach ile tüketilir.

Her logun seviyesi ParseWithSpan ile çıkarılır ve AddOrUpdate(level, 1, (_, n) => n + 1) ile sayılır. Sonuçlar OrderBy(k => k.Key) ile alfabetik yazdırılır.

Uygulama Run metodunu ekle

LogProcessor'a public static async System.Threading.Tasks.Task Run(string[] logs) ekle. İçinde System.Threading.Channels.Channel.CreateUnbounded<string>() ile kanal, new System.Collections.Concurrent.ConcurrentDictionary<string, int>() ile sözlük yarat. Logları await channel.Writer.WriteAsync(log) ile yaz, channel.Writer.Complete() çağır. await foreach (var log in channel.Reader.ReadAllAsync()) ile her logu ParseWithSpan(log.AsSpan())'den geçirip counts.AddOrUpdate(level, 1, (_, n) => n + 1) ile say. OrderBy(k => k.Key) ile yazdır. Main'de LogProcessor.Run(logs).Wait() çağır.

Örnek çözümü göster

Run metodunu ekle. Main'de LogProcessor.Run(logs).Wait();
Örnek çözüm:
using System;
using System.Linq;

static class LogProcessor
{
    public static int ParseLevel(string level)
    {
        return level switch
        {
            "INFO" => 0,
            "WARN" => 1,
            "ERROR" => 2,
            _ => -1
        };
    }

    public static int CountByLevel(string[] logs, int targetLevel)
    {
        int count = 0;
        foreach (var log in logs)
        {
            if (ParseLevel(log.Split(':')[0]) == targetLevel)
            {
                count++;
            }
        }
        return count;
    }

    public static string ParseWithSpan(ReadOnlySpan<char> log)
    {
        int colon = log.IndexOf(':');
        if (colon < 0) return "";
        return log.Slice(0, colon).ToString();
    }

    public static unsafe int UnsafeSum(int a, int b)
    {
        int* pa = &a;
        int* pb = &b;
        return *pa + *pb;
    }

    public static async System.Threading.Tasks.Task Run(string[] logs)
    {
        var channel = System.Threading.Channels.Channel.CreateUnbounded<string>();
        var counts = new System.Collections.Concurrent.ConcurrentDictionary<string, int>();
        foreach (var log in logs)
        {
            await channel.Writer.WriteAsync(log);
        }
        channel.Writer.Complete();
        await foreach (var log in channel.Reader.ReadAllAsync())
        {
            var level = ParseWithSpan(log.AsSpan());
            counts.AddOrUpdate(level, 1, (_, n) => n + 1);
        }
        foreach (var kvp in counts.OrderBy(k => k.Key))
        {
            Console.WriteLine($"{kvp.Key}: {kvp.Value}");
        }
    }
}

class Program
{
    static void Main()
    {
        Console.WriteLine("LogProcessor iskeleti hazır");
        var logs = new[] { "INFO: a", "WARN: b", "INFO: c", "ERROR: d" };
        Console.WriteLine(LogProcessor.CountByLevel(logs, 0));
        Console.WriteLine(LogProcessor.ParseWithSpan("ERROR: disk".AsSpan()));
        Console.WriteLine(LogProcessor.UnsafeSum(3, 4));
        LogProcessor.Run(logs).Wait();
    }
}

Kendini test et

Channel akış adımlarını eşleştir

CreateUnbounded

Kanalı oluştur
WriteAsync

Log üret
ReadAllAsync

Logları tüket

AddOrUpdate çağrısını tamamla

counts.____(level, 1, (_, n) => n + 1);

Cevabı göster

counts.AddOrUpdate(level, 1, (_, n) => n + 1);

Kanalı üretim sona erdi diye kapatmak için hangi metot çağrılır?
- Writer.Close()
- Writer.Finish()
- Writer.End()
- Writer.Complete()
Cevabı göster

Writer.Complete() — Doğru. Writer.Complete() yazmayı bitirir.

Channels ve concurrent