C# Academy · Lekcja

System.Threading.Channels

Twórz potoki producent-konsument za pomocą Channel, ChannelWriter i ChannelReader, aby kontrolować backpressure.

System.Threading.Channels to bezpłatna lekcja C# Academy na CoddyKit. To lekcja 2 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej C# Academy, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs C# Academy zawiera 4 lekcji w sumie.

Czym są kanały?

System.Threading.Channels (wprowadzone w .NET Core 3) udostępnia wysokowydajną, bezpieczną wątkowo kolejkę producent–konsument. W przeciwieństwie do BlockingCollection kanały są w pełni asynchroniczne i efektywne pod względem alokacji — idealnie nadają się do potoków działających w obrębie jednego procesu.

Tworzenie kanału

Kanały tworzy się za pomocą fabryki. Należy wybrać kanał nieograniczony (bez limitu) albo ograniczony (o określonej pojemności, z backpressure). Fabryka zwraca Channel<T> z elementami Writer i Reader.

using System.Threading.Channels;

// Unbounded: unlimited capacity, no backpressure
var unbounded = Channel.CreateUnbounded<string>();

// Bounded: max 100 items; writer waits when full
var bounded = Channel.CreateBounded<string>(100);

// Bounded with drop-oldest strategy:
var dropping = Channel.CreateBounded<string>(new BoundedChannelOptions(50)
{
    FullMode = BoundedChannelFullMode.DropOldest
});

Zapisywanie do kanału

Producenci zapisują elementy za pomocą WriteAsync (oczekuje, jeśli ograniczony kanał jest pełny) albo TryWrite (natychmiast zwraca false, jeśli kanał jest pełny). Zakończenie zapisywania należy zasygnalizować za pomocą Writer.Complete().

var channel = Channel.CreateUnbounded<int>();

// Producer task
var producer = Task.Run(async () =>
{
    for (int i = 0; i < 100; i++)
    {
        await channel.Writer.WriteAsync(i);
        await Task.Delay(10);
    }
    channel.Writer.Complete(); // signal no more items
});

Odczytywanie z kanału

Konsumenci odczytują dane za pomocą ReadAllAsync() (najprostsze podejście) albo ReadAsync/TryRead, jeśli potrzebują większej kontroli. ReadAllAsync kończy działanie, gdy zapisujący wywoła Complete().

// Consumer task
var consumer = Task.Run(async () =>
{
    await foreach (var item in channel.Reader.ReadAllAsync())
    {
        Console.WriteLine($"Processed: {item}");
        // Naturally paced — waits for next item
    }
    Console.WriteLine("Channel completed");
});

await Task.WhenAll(producer, consumer);

Wielu producentów

Kanały są bezpieczne wątkowo. Wielu producentów może zapisywać dane jednocześnie — blokowanie nie jest wymagane. Writer.Complete() należy wywołać dopiero po zakończeniu pracy wszystkich producentów.

var channel = Channel.CreateUnbounded<WorkItem>();

var producers = Enumerable.Range(0, 4).Select(id =>
    Task.Run(async () =>
    {
        for (int i = 0; i < 25; i++)
            await channel.Writer.WriteAsync(new WorkItem(id, i));
    }));

// Wait for all producers before completing the writer
await Task.WhenAll(producers);
channel.Writer.Complete();

Wielu konsumentów (fan-out)

Aby zrównoleglić przetwarzanie, należy uruchomić wiele zadań konsumentów korzystających z tego samego czytnika kanału. Każdy element jest dostarczany dokładnie jednemu konsumentowi (jest dzielony, a nie rozgłaszany).

var channel = Channel.CreateBounded<WorkItem>(100);

// 4 parallel consumers
var consumers = Enumerable.Range(0, 4).Select(id =>
    Task.Run(async () =>
    {
        await foreach (var item in channel.Reader.ReadAllAsync())
        {
            await ProcessItemAsync(item);
            Console.WriteLine($"Consumer {id} processed {item.Id}");
        }
    }));

await Task.WhenAll(consumers);

Wzorzec potoku

Kanały można łączyć w potok przetwarzania: każdy etap odczytuje dane z jednego kanału, przetwarza elementy i zapisuje je do następnego. Etapy działają współbieżnie, z naturalnym mechanizmem backpressure.

// Stage 1: raw data
var stage1 = Channel.CreateBounded<string>(50);
// Stage 2: parsed
var stage2 = Channel.CreateBounded<ParsedRecord>(50);
// Stage 3: enriched
var stage3 = Channel.CreateBounded<EnrichedRecord>(50);

var parse   = ParseStageAsync(stage1.Reader, stage2.Writer);
var enrich  = EnrichStageAsync(stage2.Reader, stage3.Writer);
var persist = PersistStageAsync(stage3.Reader);

await Task.WhenAll(parse, enrich, persist);

Backpressure w ograniczonych kanałach

Ograniczony kanał automatycznie stosuje backpressure: gdy kanał jest pełny, WriteAsync wstrzymuje producenta do czasu odebrania elementów przez konsumenta. Nie jest potrzebny żaden ręczny kod ograniczający tempo.

// Bounded channel: max 10 items
var channel = Channel.CreateBounded<string>(10);

// Fast producer
var producer = Task.Run(async () =>
{
    for (int i = 0; i < 1000; i++)
    {
        // WriteAsync waits when channel has 10 items
        await channel.Writer.WriteAsync($"item-{i}");
        // Producer is naturally slowed to consumer speed
    }
    channel.Writer.Complete();
});

// Slow consumer
var consumer = Task.Run(async () =>
{
    await foreach (var item in channel.Reader.ReadAllAsync())
    {
        await Task.Delay(50); // simulate slow processing
        Console.WriteLine(item);
    }
});

Obsługa błędów w kanałach

Wyjątek należy przekazać do Writer.Complete(exception), aby przekazać błędy wszystkim oczekującym czytelnikom. Czytelnicy otrzymają wyjątek przy następnym odczycie z kanału.

var channel = Channel.CreateUnbounded<int>();

var producer = Task.Run(async () =>
{
    try
    {
        for (int i = 0; i < 100; i++)
        {
            if (i == 50) throw new Exception("Producer failed at 50");
            await channel.Writer.WriteAsync(i);
        }
        channel.Writer.Complete();
    }
    catch (Exception ex)
    {
        channel.Writer.Complete(ex); // propagate to reader
    }
});

try
{
    await foreach (var item in channel.Reader.ReadAllAsync())
        Console.WriteLine(item);
}
catch (Exception ex)
{
    Console.Error.WriteLine($"Channel error: {ex.Message}");
}

Praktyczny przykład: kolejka zadań w tle

Kolejka zadań w tle korzystająca z usługi hostowanej i kanału: żądania HTTP umieszczają zadania w kolejce, a pracownik działający w tle przetwarza je pojedynczo.

public class JobQueue
{
    private readonly Channel<Func<CancellationToken, Task>> _queue
        = Channel.CreateBounded<Func<CancellationToken, Task>>(100);

    public ChannelWriter<Func<CancellationToken, Task>> Writer => _queue.Writer;
    public ChannelReader<Func<CancellationToken, Task>> Reader => _queue.Reader;
}

public class JobProcessor : BackgroundService
{
    private readonly JobQueue _queue;
    public JobProcessor(JobQueue q) => _queue = q;

    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        await foreach (var job in _queue.Reader.ReadAllAsync(ct))
            await job(ct);
    }
}

Szybkie sprawdzenie

Co się dzieje, gdy producent wywoła WriteAsync na pełnym ograniczonym kanale?

Podsumowanie: System.Threading.Channels

Najważniejsze wnioski:

Kanały udostępniają wysokowydajną, natywnie asynchroniczną kolejkę producent–konsument
Nieograniczony: bez limitu; ograniczony: o skończonej pojemności, z backpressure lub strategiami odrzucania
Wielu producentów i konsumentów może bezpiecznie korzystać z kanałów bez blokowania
ReadAllAsync() + await foreach = najczytelniejszy wzorzec konsumpcji
Kanały można łączyć w potoki na potrzeby współbieżnego przetwarzania etapowego
Należy wywołać Writer.Complete() (lub Complete(exception)), aby zasygnalizować koniec strumienia

Często zadawane pytania

Czy lekcja „System.Threading.Channels” jest bezpłatna?

Tak — pełny tekst „System.Threading.Channels” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu C# Academy, przejdź na CoddyKit PRO. Kurs C# Academy zawiera 4 lekcji w sumie.

Co nauczysz się w „System.Threading.Channels”?

Twórz potoki producent-konsument za pomocą Channel, ChannelWriter i ChannelReader, aby kontrolować backpressure. Ćwiczysz C# Academy z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.

Czy potrzebuję doświadczenia, aby zacząć C# Academy?

Nie wymagamy żadnego doświadczenia. C# Academy w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 2 z 4.

Ile czasu zajmuje lekcja „System.Threading.Channels”?

Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.

Czy mogę pisać i uruchamiać kod w tej lekcji C# Academy?

Tak. Każda lekcja C# Academy zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.

Wszystkie lekcje w tym kursie

IAsyncEnumerable i await foreach
System.Threading.Channels
ValueTask i unikanie alokacji
ConfigureAwait i kontekst synchronizacji

← Powrót do C# Academy