День 1222. #ЗаметкиНаПолях #AsyncTips
Блокирующие очереди
Задача
Необходимо создать коммуникационный канал для передачи сообщений или данных между потоками. Например, один поток может загружать данные, которые отправляются по каналу по мере загрузки; другие потоки на стороне получения получают эти данные и обрабатывают их.
Решение
Тип .NET BlockingCollection<T> проектировался для создания таких коммуникационных каналов. По умолчанию BlockingCollection<T> работает в режиме блокирующей очереди и предоставляет поведение «первым зашел, первым вышел».
Блокирующая очередь должна совместно использоваться несколькими потоками, и обычно определяется как приватное поле, доступное только для чтения:
private readonly BlockingCollection<int> _bq =
new BlockingCollection<int>();
Обычно поток делает что-то одно: либо добавляет элементы в коллекцию, либо удаляет элементы. Потоки, добавляющие элементы, называются потоками-производителями, а потоки, удаляющие элементы, называются потоками-потребителями.
Потоки-производители могут добавлять элементы вызовами Add, а когда поток-производитель завершится (когда будут добавлены все элементы), он может завершить коллекцию вызовом CompleteAdding. Тем самым он уведомляет коллекцию о том, что элементы далее добавляться не будут, а коллекция может сообщить своим потребителям, что элементов больше не будет.
В следующем простом примере производитель добавляет два элемента, а потом помечает коллекцию как завершенную:
_bq.Add(7);
_bq.Add(13);
_bq.CompleteAdding();
Потоки-потребители обычно выполняются в цикле, ожидая следующего элемента и выполняя его последующую обработку. Если выделить код производителя в отдельный поток (например, вызовом Task.Run), то эти элементы можно будет потреблять следующим образом:
// Выводит "7", затем "13".
foreach (int item in _bq.GetConsumingEnumerable())
Console.WriteLine(item);
Если потребителей должно быть несколько, GetConsumingEnumerable может вызываться из нескольких потоков одновременно. Тем не менее каждый элемент передается только одному из этих потоков. При завершении коллекции завершается и перечисляемый объект.
Во всех приведенных примерах GetConsumingEnumerable используется для потоков-потребителей; это самая распространённая ситуация. Но существует и метод Take, который позволяет потребителю получить только один элемент (вместо потребления всех элементов в цикле).
При использовании таких коммуникационных каналов необходимо подумать о том, что произойдет, если производители работают быстрее потребителей. Если элементы производятся быстрее, чем потребляются, возможно, придётся применить регулировку очереди.
Блокирующие очереди хорошо работают при наличии отдельного потока (например, из пула потоков), действующего как производитель или потребитель. Они не настолько хороши, если вы хотите обращаться к коммуникационному каналу асинхронно — например, если UI-поток должен действовать в режиме потребителя. Если вы вводите в своё приложение подобный коммуникационный канал, подумайте о переходе на библиотеку TPL Dataflow. Во многих случаях решение с использованием TPL Dataflow проще самостоятельного построения коммуникационных каналов и фоновых потоков.
Тип BufferBlock<T> из TPL Dataflow может работать как блокирующая очередь, к тому же TPL Dataflow позволяет построить конвейер или сеть для обработки. Впрочем, во многих простых случаях обычные блокирующие очереди (например, BlockingCollection<T>) станут более подходящим вариантом при проектировании.
Источник: Стивен Клири “Конкурентность в C#”. 2-е межд. изд. — СПб.: Питер, 2020. Глава 9.
