Channel: produttore–consumatore ( Channel)

1. Introduzione

Oggi saliamo di livello! È ora di conoscere uno strumento tutto speciale — Channel. Questa cosa è stata creata apposta per le applicazioni asincrone moderne in .NET: dove i blocchi tradizionali o non bastano, o ti rallentano parecchio.

Ti aspetta il pattern "Produttore-Consumatore" (Producer-Consumer), che non passa mai di moda dagli anni '60 del XX secolo. Creerai una semplice "catena" asincrona dove alcuni thread o task producono qualcosa (per esempio scaricano file, conteggiano numeri, aspettano eventi) e altri li consumano (per esempio salvano, scrivono su DB, aggiornano l'UI).

Perché è nato Channel?

1. Il pattern "Produttore-Consumatore" da sempre si risolveva con code: il produttore mette task nella coda, il consumatore li prende. Ma! BlockingCollection<T>, le code basate su ConcurrentQueue<T>, persino la sincronizzazione manuale con lock — tutto ciò non è asincrono. Quindi il thread può solo bloccarsi in attesa di dati, invece di restituire il controllo allo scheduler di async/await.
2. L'asincronia in .NET non è solo una parola alla moda del venerdì, ma il fondamento dell'architettura moderna. Bloccare thread per aspettare elementi è costoso e inefficiente. Serve poter aspettare l'arrivo dei dati senza blocking — ed è proprio quello che risolve Channel.
3. Flessibilità: con i canali puoi costruire pipeline complesse, separare la logica dei thread, aggiungere step intermedi e bilanciamento del carico — e tutto questo senza impazzire con la sincronizzazione a basso livello.

Cos'è Channel? (Analogia e architettura)

Immagina di avere una staffetta (o un nastro trasportatore) su cui passare oggetti da un punto all'altro, senza che le persone debbano incontrarsi faccia a faccia. L'importante è che la "staffetta" non si perda per strada.

Channel è uno strumento integrato in .NET per la trasmissione asincrona dei dati tra diversi task, thread o parti del programma. Implementa una coda asincrona che supporta l'"attesa" sia per l'inserimento che per l'estrazione degli elementi.

• Il produttore mette elementi nel canale (per esempio richieste da processare);
• Il consumatore prende gli elementi — e il gioco è fatto!

2. La classe Channel<T> e la sua struttura

Tutto parte dallo namespace:

using System.Threading.Channels;

Diversamente dalle collezioni tradizionali, Channel è una factory che crea oggetti speciali per trasferire dati.

Tipi principali:

• ChannelWriter<T> — lo "scrittore" (producer). Fa solo l'inserimento degli elementi.
• ChannelReader<T> — il "lettore" (consumer). Fa solo l'estrazione degli elementi.
• Il canale (Channel) separa le responsabilità: lo scrittore non sa nulla del lettore e viceversa.

In .NET esistono diverse implementazioni di canale, ognuna con le sue caratteristiche: unbounded (senza limiti), bounded (con limite sul numero di elementi), single-producer-single-consumer (SPSC), multi-producer-multi-consumer (MPMC), ecc. Partiremo dalla più universale.

Esempio semplice: una coda asincrona di task

using System;
using System.Threading.Channels;
using System.Threading.Tasks;

class Program
{
    static async Task Main()
    {
        // Creiamo un canale senza limite di dimensione
        var channel = Channel.CreateUnbounded<int>();

        // Task produttore
        var producer = Task.Run(async () =>
        {
            for (int i = 0; i < 10; i++)
            {
                Console.WriteLine($"Produttore: Inserisce {i} nel canale");
                await channel.Writer.WriteAsync(i); // Scrittura asincrona!
                await Task.Delay(100); // Simulazione lavoro
            }
            channel.Writer.Complete(); // Diciamo che non scriveremo più
        });

        // Task consumatore
        var consumer = Task.Run(async () =>
        {
            await foreach (var item in channel.Reader.ReadAllAsync())
            {
                Console.WriteLine($"Consumatore: Ha ricevuto {item} dal canale");
                await Task.Delay(200); // Simulazione elaborazione
            }
            Console.WriteLine("Consumatore: Il canale si è chiuso");
        });

        await Task.WhenAll(producer, consumer);
    }
}

Cosa succede qui?

• Channel.CreateUnbounded<int>() — creiamo un canale senza limiti di lunghezza.
• Il produttore scrive numeri da 0 a 9 nel canale usando WriteAsync.
• Dopo aver finito di scrivere si chiama Complete() — segnale "Non ci saranno altri elementi!".
• Il consumatore scorre tutti gli elementi con ReadAllAsync() (anch'esso asincrono!), finché il canale non si chiude.
• I ritardi (Task.Delay) simulano lavoro reale: vedi che i numeri vengono scritti più velocemente di quanto vengano letti.

3. Perché tutto ciò funziona in modo asincrono?

Le code bloccanti tradizionali (per esempio BlockingCollection o quelle protette con lock) possono solo bloccare un thread. Questo significa perdere risorse preziose se hai molti task o vuoi massima performance.

Con i canali:

• Se il produttore è più veloce, il canale accumula elementi (limitato solo dalla memoria o dal massimo se specificato).
• Se il consumatore è più veloce, aspetterà che qualcosa arrivi (e non bloccherà per sempre il thread, ma lo restituisce allo scheduler).

Questo è l'ideale quando non sai a priori chi sarà più veloce — i produttori o i consumatori.

Applicazioni nella vita reale

• Logging asincrono: scrivere messaggi su file/DB in un thread separato;
• Elaborazione di richieste web: un task scarica batch di pagine, un altro le analizza;
• Scansione e indicizzazione di cartelle: alcuni task attraversano il file system, altri calcolano statistiche sui file;
• Pipeline complesse di elaborazione dati: ad esempio in ETL (Extract–Transform–Load) un passo trasforma la materia prima in semilavorati, un altro li trasforma in prodotto finito.

4. Canale limitato (Bounded Channel)

I canali "senza limiti" sono divertenti, ma la memoria non è infinita (anche se il tuo PC ti sembra enorme).

Un canale limitato (bounded) permette di impostare il numero massimo di elementi che possono stare contemporaneamente dentro. Se il canale è pieno — il produttore aspetta finché il consumatore non ne estrae qualcuno.

Esempio:

var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(3)
{
    FullMode = BoundedChannelFullMode.Wait // (default) - aspettare che si liberi spazio
});

Qui solo tre elementi possono stare nel canale contemporaneamente. Se il produttore prova a scriverne un quarto — aspetterà.

Più produttori e consumatori

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

// 2 produttori
for (int producerId = 0; producerId < 2; producerId++)
{
    Task.Run(async () =>
    {
        for (int i = 0; i < 5; i++)
        {
            int value = producerId * 100 + i;
            Console.WriteLine($"Produttore {producerId}: Inserisce {value}");
            await channel.Writer.WriteAsync(value);
            await Task.Delay(50);
        }
        // Ogni produttore chiama Complete() — pericoloso!
    });
}
// Trucco: Complete() deve essere chiamato solo una volta, quando TUTTI i produttori hanno finito.
// Nella realtà si usa un contatore di task o Task.WhenAll.
Task.Run(async () =>
{
    await foreach (var item in channel.Reader.ReadAllAsync())
    {
        Console.WriteLine($"Il consumatore ha ricevuto {item}");
        await Task.Delay(100);
    }
});

Attenzione! Il canale deve essere chiuso (con Complete()) solo dopo che tutti i produttori hanno terminato. Altrimenti altri potrebbero provare a scrivere mentre il canale è già chiuso! Nella pratica si usa di solito un contatore dei task o Task.WhenAll.

5. Pratica: Elaborazione di immagini tramite canale

Complichiamo un po' il caso! Immaginiamo di avere una cartella con immagini. Un task cerca le immagini e mette i loro percorsi nel canale, un altro prende il percorso e fa qualcosa di utile con il file (per esempio calcola la dimensione o converte).

Nota: Per semplicità l'esempio lavorerà solo con i nomi dei file (senza manipolare veramente le immagini), ma il concetto è identico.

using System;
using System.IO;
using System.Threading.Channels;
using System.Threading.Tasks;

class Program
{
    static async Task Main()
    {
        var channel = Channel.CreateBounded<string>(5);

        // Produttore: cerca file .jpg nella cartella
        var producer = Task.Run(async () =>
        {
            foreach (var file in Directory.EnumerateFiles(@"images", "*.jpg"))
            {
                await channel.Writer.WriteAsync(file);
                Console.WriteLine($"Aggiunto in coda: {file}");
                await Task.Delay(50); // simuliamo ritardo della ricerca
            }
            channel.Writer.Complete(); // fine della coda
        });

        // Consumatore: legge e "elabora" i file
        var consumer = Task.Run(async () =>
        {
            await foreach (var file in channel.Reader.ReadAllAsync())
            {
                Console.WriteLine($"Elaborazione file: {file}");
                await Task.Delay(200); // simulazione elaborazione
            }
            Console.WriteLine("Tutte le immagini sono state elaborate!");
        });

        await Task.WhenAll(producer, consumer);
    }
}

6. Configurare Channel: opzioni e dettagli

I canali si possono configurare tramite opzioni alla creazione — ecco i parametri principali per i canali limitati:

Capacity

SingleWriter

SingleReader

FullMode

Esempio con opzioni:

var options = new BoundedChannelOptions(10)
{
    SingleWriter = false,
    SingleReader = true,
    FullMode = BoundedChannelFullMode.Wait
};
var channel = Channel.CreateBounded<string>(options);

7. Metodi asincroni: ReadAsync, WriteAsync, ReadAllAsync

Perché async è così importante?

I metodi WriteAsync e ReadAsync non bloccano il thread! Se non c'è nulla da leggere — il task viene messo in pausa, liberando il thread per altri lavori. Questo è cruciale per applicazioni server e UI, dove un blocco in più può causare freeze.

ReadAllAsync — la comodità del C# moderno

Si può iterare in modo asincrono:

await foreach (var item in channel.Reader.ReadAllAsync())
{
    // Lavoriamo con item
}

Channel<T> vs collezioni thread-safe: qual è la differenza?

ConcurrentQueue<T>/BlockingCollection<T> vanno bene per scenari basati su thread, ma non sono adatte per la pura asincronia (scenari con await).

Channel<T> è pensato proprio per pipeline asincrone. Dal punto di vista della sicurezza dei thread, entrambe le soluzioni funzionano bene, ma i canali offrono flessibilità e integrazione con le feature moderne di C# (come IAsyncEnumerable e simili).

8. Errori e trappole comuni

Non dimenticare di chiamare Complete() sul writer quando hai aggiunto tutti gli elementi! Altrimenti il consumatore resterà appeso in attesa di nuovi elementi per sempre.

Non chiamare Complete() più volte se ci sono molti writer — fallo solo dopo che tutti i produttori hanno finito.

Dopo la chiusura del canale non si possono più scrivere elementi, ma si possono leggere quelli rimanenti.

Race condition durante scritture contemporanee: se il canale è chiuso e qualcuno prova comunque a scrivere — otterrai un'eccezione.

2

Compito

C# SELF, livello 62, lezione 3

Bloccato

Creazione di un semplice canale per trasmettere numeri

Creazione di un semplice canale per trasmettere numeri

Apri