C# Academy · Lezione

Producer/consumer con Channels (panoramica)

Modellate code producer/consumer come Channels usando BlockingCollection : buffer con capacità limitata (backpressure), più consumer e completamento ordinato.

Producer/consumer con Channels (panoramica) è una lezione C# Academy gratuita su CoddyKit. Questa è la lezione 2 di 3. 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 3 lezioni in totale.

Concetto e obiettivi

Obiettivo: costruisca una pipeline sicura produttore/consumatore.

Idea del canale: una coda tra produttori e consumatori
Una capacità limitata fornisce backpressure
Supportare più consumatori
Usare CompleteAdding per un arresto ordinato

Un produttore e un consumatore

Flusso di base: i produttori aggiungono elementi con Add, i consumatori li enumerano con GetConsumingEnumerable, quindi si usa CompleteAdding per terminare.

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

public class Program
{
  public static void Main(string[] args)
  {
    // Unbounded by default; simple demo
    BlockingCollection<int> queue = new BlockingCollection<int>();

    // Producer
    Task producer = Task.Run(() =>
    {
      for (int i = 1; i <= 5; i++)
      {
        queue.Add(i);                 // enqueue
        Console.WriteLine("Produced " + i);
        Thread.Sleep(50);             // simulate work
      }
      queue.CompleteAdding();         // signal no more items
    });

    // Consumer
    Task consumer = Task.Run(() =>
    {
      foreach (int x in queue.GetConsumingEnumerable())
      {
        Console.WriteLine("Consumed " + x);
        Thread.Sleep(80);             // simulate processing
      }
      Console.WriteLine("Consumer done");
    });

    Task.WaitAll(producer, consumer);
  }
}

Dimostrazione del backpressure

Le code a capacità limitata applicano il backpressure: quando sono piene, Add si blocca finché i consumatori non creano spazio.

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

public class Program
{
  public static void Main(string[] args)
  {
    // Bounded capacity introduces backpressure
    BlockingCollection<int> queue = new BlockingCollection<int>(boundedCapacity: 2);

    Task producer = Task.Run(() =>
    {
      for (int i = 1; i <= 5; i++)
      {
        queue.Add(i); // blocks when the buffer is full
        Console.WriteLine("Produced " + i + " (count=" + queue.Count + ")");
      }
      queue.CompleteAdding();
    });

    Task consumer = Task.Run(() =>
    {
      foreach (int x in queue.GetConsumingEnumerable())
      {
        Console.WriteLine("Consumed " + x);
        Thread.Sleep(120); // slower consumer -> producer will block sometimes
      }
    });

    Task.WaitAll(producer, consumer);
  }
}

Consumatori fan-out

Più consumatori competono per gli elementi (fan-out). Il lavoro viene distribuito automaticamente tra i thread.

using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

public class Program
{
  public static void Main(string[] args)
  {
    BlockingCollection<int> queue = new BlockingCollection<int>(3);

    Task prod = Task.Run(() =>
    {
      for (int i = 1; i <= 8; i++)
      {
        queue.Add(i);
        Console.WriteLine("Produced " + i);
        Thread.Sleep(30);
      }
      queue.CompleteAdding();
    });

    // Start 2 consumers that compete for items
    Task[] consumers = Enumerable.Range(1, 2).Select(id => Task.Run(() =>
    {
      foreach (int x in queue.GetConsumingEnumerable())
      {
        Console.WriteLine("C" + id + " got " + x);
        Thread.Sleep(100);
      }
      Console.WriteLine("C" + id + " done");
    })).ToArray();

    Task.WaitAll(consumers.Concat(new[] { prod }).ToArray());
  }
}

Completamento ordinato

Usi CompleteAdding e GetConsumingEnumerable per terminare correttamente quando la coda si svuota.

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

public class Program
{
  static void Producer(BlockingCollection<string> q)
  {
    string[] lines = { "A", "B", "C", "D" };
    foreach (var s in lines)
    {
      q.Add(s);
      Console.WriteLine("Produced " + s);
    }
    q.CompleteAdding(); // signal completion
  }

  static void Consumer(BlockingCollection<string> q)
  {
    try
    {
      foreach (var item in q.GetConsumingEnumerable())
      {
        Console.WriteLine("Consumed " + item);
        Thread.Sleep(60);
      }
      Console.WriteLine("Consumer finished all items");
    }
    catch (InvalidOperationException)
    {
      // Thrown if taken after completion and empty; avoided by foreach above.
    }
  }

  public static void Main(string[] args)
  {
    var queue = new BlockingCollection<string>(2);
    Task p = Task.Run(() => Producer(queue));
    Task c = Task.Run(() => Consumer(queue));
    Task.WaitAll(p, c);
  }
}

Suggerimenti e problemi comuni

Suggerimenti:

Preferisca buffer a capacità limitata per evitare la crescita della memoria.
Usi GetConsumingEnumerable per evitare race condition durante il completamento.
Mantenga le unità di lavoro piccole e indipendenti; eviti lo stato condiviso.
Registri e gestisca gli errori all'interno dei consumatori per evitare interruzioni silenziose.

Scopo di CompleteAdding

Verifica rapida: che cosa accade quando si chiama CompleteAdding() su una BlockingCollection?

Riepilogo

Riepilogo: usi una BlockingCollection a capacità limitata per simulare i canali, distribuisca il lavoro a più consumatori e usi CompleteAdding per un arresto ordinato.

Domande Frequenti

La lezione «Producer/consumer con Channels (panoramica)» è gratuita?

Sì — il testo completo di «Producer/consumer con Channels (panoramica)» è 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 3 lezioni in totale.

Cosa imparerò in «Producer/consumer con Channels (panoramica)»?

Modellate code producer/consumer come Channels usando BlockingCollection : buffer con capacità limitata (backpressure), più consumer e completamento ordinato. 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 3.

Quanto tempo richiede la lezione «Producer/consumer con Channels (panoramica)»?

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

Parallel.ForEach, PLINQ
Producer/consumer con Channels (panoramica)
Compromessi tra throughput e latenza

← Torna a C# Academy