C# Academy · Lekcja

Producent i konsument z użyciem Channels (omówienie)

Proszę modelować kolejki producent/konsument podobne do Channels za pomocą BlockingCollection : ograniczone bufory (backpressure), wielu konsumentów i poprawne kończenie pracy.

Producent i konsument z użyciem Channels (omówienie) to bezpłatna lekcja C# Academy na CoddyKit. To lekcja 2 z 3. 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 3 lekcji w sumie.

Pojęcie i cele

Cel: Zbudowanie bezpiecznego potoku producent–konsument.

Idea kanału: kolejka między producentami a konsumentami
Ograniczona pojemność zapewnia mechanizm backpressure
Obsługa wielu konsumentów
Użycie CompleteAdding do poprawnego zamykania

Jeden producent i jeden konsument

Podstawowy przepływ: producenci wywołują Add, konsumenci iterują po GetConsumingEnumerable, a następnie wywoływana jest metoda CompleteAdding, aby zakończyć pracę.

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

Przykład backpressure

Ograniczone kolejki stosują mechanizm backpressure: gdy są pełne, metoda Add blokuje działanie do czasu zwolnienia miejsca przez konsumentów.

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

Konsumenci w modelu fan-out

Wielu konsumentów rywalizuje o elementy (fan-out). Praca jest automatycznie rozdzielana między wątki.

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

Poprawne zakończenie

Należy użyć CompleteAdding i GetConsumingEnumerable, aby poprawnie zakończyć pracę po opróżnieniu kolejki.

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

Wskazówki i pułapki

Wskazówki:

Należy preferować ograniczone bufory, aby zapobiegać wzrostowi zużycia pamięci.
Należy używać GetConsumingEnumerable, aby uniknąć wyścigów podczas kończenia pracy.
Elementy pracy powinny być małe i niezależne; należy unikać współdzielonego stanu.
Należy rejestrować błędy i obsługiwać je wewnątrz konsumentów, aby uniknąć niejawnego zatrzymania pracy.

Cel CompleteAdding

Szybkie sprawdzenie: Co dzieje się po wywołaniu CompleteAdding() na obiekcie BlockingCollection?

Podsumowanie

Podsumowanie: Należy używać ograniczonej kolekcji BlockingCollection do emulowania kanałów, rozdzielać pracę między wielu konsumentów i wywoływać CompleteAdding w celu zapewnienia łagodnego zamknięcia.

Często zadawane pytania

Czy lekcja „Producent i konsument z użyciem Channels (omówienie)” jest bezpłatna?

Tak — pełny tekst „Producent i konsument z użyciem Channels (omówienie)” 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 3 lekcji w sumie.

Co nauczysz się w „Producent i konsument z użyciem Channels (omówienie)”?

Proszę modelować kolejki producent/konsument podobne do Channels za pomocą BlockingCollection : ograniczone bufory (backpressure), wielu konsumentów i poprawne kończenie pracy. Ć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 3.

Ile czasu zajmuje lekcja „Producent i konsument z użyciem Channels (omówienie)”?

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

Parallel.ForEach i PLINQ
Producent i konsument z użyciem Channels (omówienie)
Kompromis między przepustowością a opóźnieniem

← Powrót do C# Academy