День 1246. #ЗаметкиНаПолях #AsyncTips
Асинхронные очереди

Задача
Требуется коммуникационный канал для передачи сообщений или данных из одной части кода в другую по принципу FIFO без блокирования потоков.
Например, один фрагмент кода может загружать данные, которые отправляются по каналу по мере загрузки; при этом UI-поток получает данные и выводит их.

Решение
Требуется очередь с асинхронным API. В базовом фреймворке .NET такого типа нет, но в NuGet есть пара возможных решений.

1. System.Threading.Channels
Библиотека для асинхронных коллекций «производитель/потребитель» с акцентом на быстродействие. Производители записывают элементы в канал вызовом WriteAsync, а когда завершают производство элементов, один из них вызывает Complete для уведомления канала о том, что в дальнейшем элементов больше не будет:

var queue = Channel.CreateUnbounded<int>();
// Код-производитель
var writer = queue.Writer;
await writer.WriteAsync(7);
await writer.WriteAsync(13);
writer.Complete();

// Код-потребитель
// Выводит "7", затем "13".
var reader = queue.Reader;
await foreach (int value in reader.ReadAllAsync())
  Console.WriteLine(value);

Код-потребитель использует асинхронные потоки.

2. System.Threading.Tasks.Dataflow
BufferBlock<T> из библиотеки TPL Dataflow имеет много общего с каналом:

var queue = new BufferBlock<int>();
// Код-производитель
await queue.SendAsync(7);
await queue.SendAsync(13);
queue.Complete();

// Код-потребитель.
// Выводит "7", затем "13".
while (await queue.OutputAvailableAsync())
  Console.WriteLine(await queue.ReceiveAsync());

Код-потребитель использует метод OutputAvailableAsync, который на самом деле полезен только с одним потребителем. Если потребителей несколько, может случиться, что OutputAvailableAsync вернет true для нескольких потребителей, хотя элемент только один. Если очередь завершена, то ReceiveAsync выдаст исключение InvalidOperationException. Таким образом, для нескольких потребителей код будет выглядеть так:

while (true)
{
  int item;
  try
  {
    item = await queue.ReceiveAsync();
  }
  catch (InvalidOperationException)
  {
    break;
  }
  Console.WriteLine(item);
}

Библиотека Channels больше подходит для асинхронных очередей «производитель/потребитель» там, где это возможно. Помимо регулировки для случаев, когда производители работают быстрее потребителей, поддерживаются несколько режимов выборки (об этом в следующих постах). Однако, если логика вашего приложения может быть выражена в виде «конвейера», через который проходят данные, TPL Dataflow может быть более логичным кандидатом.

См. также
- очереди «производитель/потребитель» с блокирующей семантикой вместо асинхронной.

Источник: Стивен Клири “Конкурентность в C#”. 2-е межд. изд. — СПб.: Питер, 2020. Глава 9.