C# Academy · Lezione

System.Threading.Channels

Costruisca pipeline produttore-consumatore usando Channel, ChannelWriter e ChannelReader per controllare la contropressione.

System.Threading.Channels è una lezione C# Academy gratuita su CoddyKit. Questa è la lezione 2 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento C# Academy, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso C# Academy include 4 lezioni in totale.

Cosa sono i canali?

System.Threading.Channels (introdotto in .NET Core 3) fornisce una coda producer-consumer ad alte prestazioni e thread-safe. A differenza di BlockingCollection, i canali sono completamente asincroni e ottimizzati nell'uso delle allocazioni: sono ideali per le pipeline all'interno del processo.

Creazione di un canale

I canali vengono creati tramite una factory. Scelga un canale unbounded (senza limite) o bounded (con capacità limitata e backpressure). La factory restituisce un Channel<T> con un Writer e un 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
});

Scrittura in un canale

I produttori scrivono gli elementi utilizzando WriteAsync (attende se il canale bounded è pieno) oppure TryWrite (restituisce immediatamente false se è pieno). Segnali il completamento con 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
});

Lettura da un canale

I consumatori leggono con ReadAllAsync() (l'approccio più semplice) oppure con ReadAsync/TryRead per un maggiore controllo. ReadAllAsync termina quando il writer chiama 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);

Più produttori

I canali sono thread-safe. Più produttori possono scrivere contemporaneamente, senza bisogno di locking. Chiami Writer.Complete() solo dopo che tutti i produttori hanno terminato.

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

Più consumatori (fan-out)

Esegua più attività di consumo sullo stesso reader del canale per parallelizzare l'elaborazione. Ogni elemento viene consegnato esattamente a un consumatore (partizionato, non trasmesso a tutti).

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

Pattern pipeline

Colleghi i canali in una pipeline di elaborazione: ogni fase legge da un canale, elabora gli elementi e scrive nel successivo. Le fasi vengono eseguite contemporaneamente, con backpressure naturale.

// 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 con canali bounded

Un canale bounded applica automaticamente la backpressure: quando il canale è pieno, WriteAsync sospende il produttore finché il consumatore non estrae alcuni elementi. Non è necessario scrivere manualmente codice di limitazione della frequenza.

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

Gestione degli errori nei canali

Passi un'eccezione a Writer.Complete(exception) per propagare gli errori a tutti i lettori in attesa. I lettori ricevono l'eccezione alla lettura successiva del canale.

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

Caso reale: coda di lavori in background

Una coda di lavori in background che utilizza un hosted service e un canale: le richieste HTTP accodano i lavori e un worker in background li elabora uno alla volta.

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

Controllo rapido

Che cosa accade quando un produttore chiama WriteAsync su un canale bounded pieno?

Riepilogo: System.Threading.Channels

Punti chiave:

I canali forniscono una coda producer-consumer ad alte prestazioni e nativamente asincrona
Unbounded: nessun limite; Bounded: capacità finita con strategie di backpressure o di scarto
Più produttori e consumatori operano in sicurezza senza locking
ReadAllAsync() + await foreach = il modello di consumo più semplice
Collegare i canali in pipeline per un'elaborazione concorrente organizzata per fasi
Chiamare Writer.Complete() (o Complete(exception)) per segnalare la fine del flusso

Domande Frequenti

La lezione «System.Threading.Channels» è gratuita?

Sì — il testo completo di «System.Threading.Channels» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso C# Academy, passa a CoddyKit PRO. Il corso C# Academy include 4 lezioni in totale.

Cosa imparerò in «System.Threading.Channels»?

Costruisca pipeline produttore-consumatore usando Channel, ChannelWriter e ChannelReader per controllare la contropressione. Eserciti C# Academy con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.

Ho bisogno di esperienza per iniziare C# Academy?

Non è richiesta alcuna esperienza precedente. C# Academy su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 2 di 4.

Quanto tempo richiede la lezione «System.Threading.Channels»?

La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.

Posso scrivere ed eseguire codice in questa lezione C# Academy?

Sì. Ogni lezione C# Academy include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.

Tutte le lezioni di questo corso

IAsyncEnumerable e await foreach
System.Threading.Channels
ValueTask ed evitare allocazioni
ConfigureAwait e contesto di sincronizzazione

← Torna a C# Academy