C# Academy · Урок

Производитель и потребитель с каналами (обзор)

«Производитель и потребитель с каналами (обзор)» — бесплатный урок C# Academy на CoddyKit. Это урок 2 из 3. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения C# Academy, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс C# Academy содержит 3 уроков всего.

Понятие и цели

Цель: построить безопасный конвейер производителей и потребителей.

Идея канала: очередь между производителями и потребителями
Ограниченная вместимость создаёт обратное давление
Поддерживать несколько потребителей
Использовать CompleteAdding для корректного завершения

Один производитель и один потребитель

Базовый поток: производители вызывают Add, потребители перебирают элементы через GetConsumingEnumerable, затем вызывается CompleteAdding для завершения.

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

Демонстрация обратного давления

Ограниченные очереди создают обратное давление: когда очередь заполнена, Add блокируется, пока потребители не освободят место.

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

Распределение между потребителями

Несколько потребителей конкурируют за элементы. Работа автоматически распределяется между потоками.

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

Корректное завершение

Используйте CompleteAdding и GetConsumingEnumerable, чтобы корректно завершить работу после опустошения очереди.

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

Советы и подводные камни

Советы:

Предпочитайте ограниченные буферы, чтобы предотвратить рост потребления памяти.
Используйте GetConsumingEnumerable, чтобы избежать состязаний при завершении.
Делайте элементы работы небольшими и независимыми; избегайте общего состояния.
Записывайте ошибки в журнал и обрабатывайте их внутри потребителей, чтобы избежать незаметного прекращения работы.

Назначение CompleteAdding

Быстрая проверка: что происходит при вызове CompleteAdding() для BlockingCollection?

Итоги

Итоги: используйте ограниченную BlockingCollection для имитации каналов, распределяйте работу между несколькими потребителями и вызывайте CompleteAdding для корректного завершения.

Часто задаваемые вопросы

Урок «Производитель и потребитель с каналами (обзор)» бесплатный?

Да — полный текст урока «Производитель и потребитель с каналами (обзор)» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс C# Academy, подпишись на CoddyKit PRO. Курс C# Academy содержит 3 уроков всего.

Чему я научусь в уроке «Производитель и потребитель с каналами (обзор)»?

Моделируйте очереди производителя и потребителя наподобие Channels с помощью BlockingCollection : ограниченные буферы (обратное давление), несколько потребителей и корректное завершение. Ты практикуешь C# Academy с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.

Нужен ли мне опыт, чтобы начать C# Academy?

Предыдущий опыт не требуется. C# Academy на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 2 из 3.

Сколько времени занимает урок «Производитель и потребитель с каналами (обзор)»?

Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.

Можно ли писать и запускать код в этом уроке C# Academy?

Да. Каждый урок C# Academy включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.

Все уроки этого курса

Parallel.ForEach и PLINQ
Производитель и потребитель с каналами (обзор)
Компромиссы между пропускной способностью и задержкой

← Назад к C# Academy