C# Academy · Lektion

System.Threading.Channels

Erstellen Sie Producer-Consumer-Pipelines mit Channel, ChannelWriter und ChannelReader zur Steuerung von Backpressure.

System.Threading.Channels ist eine kostenlose C# Academy-Lektion auf CoddyKit. Dies ist Lektion 2 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des C# Academy-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der C# Academy-Kurs umfasst insgesamt 4 Lektionen.

Was sind Channels?

System.Threading.Channels (eingeführt in .NET Core 3) stellt eine performante, threadsichere Producer-Consumer-Warteschlange bereit. Im Gegensatz zu BlockingCollection sind Channels vollständig asynchron und speichereffizient — ideal für In-Process-Pipelines.

Einen Channel erstellen

Channels werden mit einer Factory erstellt. Wählen Sie unbounded (ohne Limit) oder bounded (begrenzte Kapazität mit Backpressure). Die Factory gibt einen Channel<T> mit einem Writer und einem Reader zurück.

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
});

In einen Channel schreiben

Produzenten schreiben Elemente mit WriteAsync (wartet, wenn der begrenzte Channel voll ist) oder TryWrite (gibt bei voller Kapazität sofort false zurück). Signalisieren Sie den Abschluss mit 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
});

Aus einem Channel lesen

Konsumenten lesen mit ReadAllAsync() (der einfachste Ansatz) oder mit ReadAsync/TryRead, wenn Sie mehr Kontrolle benötigen. ReadAllAsync ist abgeschlossen, sobald der Writer Complete() aufruft.

// 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);

Mehrere Produzenten

Channels sind threadsicher. Mehrere Produzenten können gleichzeitig schreiben — eine Sperrung ist nicht erforderlich. Rufen Sie Writer.Complete() erst auf, wenn alle Produzenten fertig sind.

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();

Mehrere Konsumenten (Fan-out)

Führen Sie mehrere Konsumenten-Tasks auf demselben Channel-Reader aus, um die Verarbeitung zu parallelisieren. Jedes Element wird genau einem Konsumenten zugestellt (partitioniert, nicht als Broadcast).

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);

Pipeline-Muster

Verketten Sie Channels zu einer Verarbeitungspipeline: Jede Stufe liest aus einem Channel, verarbeitet die Elemente und schreibt in den nächsten. Die Stufen laufen gleichzeitig und sorgen für natürliche 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 mit begrenzten Channels

Ein begrenzter Channel wendet automatisch Backpressure an: Wenn der Channel voll ist, hält WriteAsync den Produzenten an, bis der Konsument Elemente entnommen hat. Es ist kein manueller Throttling-Code erforderlich.

// 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);
    }
});

Fehlerbehandlung in Channels

Übergeben Sie eine Ausnahme an Writer.Complete(exception), um Fehler an alle wartenden Reader weiterzugeben. Reader erhalten die Ausnahme beim nächsten Lesevorgang aus dem Channel.

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}");
}

Praxisbeispiel: Warteschlange für Hintergrundaufgaben

Eine Warteschlange für Hintergrundaufgaben mit einem Hosted Service und einem Channel: HTTP-Anfragen stellen Aufgaben in die Warteschlange; ein Hintergrund-Worker verarbeitet sie nacheinander.

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);
    }
}

Kurzer Test

Was geschieht, wenn ein Produzent WriteAsync auf einem vollen begrenzten Channel aufruft?

Zusammenfassung: System.Threading.Channels

Wichtigste Erkenntnisse:

Channels stellen eine performante, nativ asynchrone Producer-Consumer-Warteschlange bereit
Unbounded: kein Limit; Bounded: begrenzte Kapazität mit Backpressure- oder Verwerfungsstrategien
Mehrere Produzenten und Konsumenten sind ohne Sperren sicher verwendbar
ReadAllAsync() + await foreach = sauberstes Muster zum Konsumieren
Verketten Sie Channels zu Pipelines für eine gleichzeitige Verarbeitung in Stufen
Rufen Sie Writer.Complete() (oder Complete(exception)) auf, um das Ende des Streams zu signalisieren

Häufig gestellte Fragen

Ist die Lektion „System.Threading.Channels“ kostenlos?

Ja — der vollständige Text von „System.Threading.Channels“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des C# Academy-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der C# Academy-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „System.Threading.Channels“?

Erstellen Sie Producer-Consumer-Pipelines mit Channel, ChannelWriter und ChannelReader zur Steuerung von Backpressure. Du übst C# Academy mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um C# Academy zu starten?

Keine Vorkenntnisse erforderlich. C# Academy auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 2 von 4.

Wie lange dauert die Lektion „System.Threading.Channels“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser C# Academy-Lektion Code schreiben und ausführen?

Ja. Jede C# Academy-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

IAsyncEnumerable und await foreach
System.Threading.Channels
ValueTask und Vermeidung von Allokationen
ConfigureAwait und SynchronizationContext

← Zurück zu C# Academy