.NET Разработчик: “День 2783. #ЗаметкиНаПолях Паттерн «Производитель-потребитель» c Syste…”

День 2783. #ЗаметкиНаПолях
Паттерн «Производитель-потребитель» c System.Threading.Channels. Окончание
Начало
Продолжение

Настройка и запуск нескольких потребителей
Зарегистрируйте канал как синглтон, чтобы производитель и потребитель использовали один и тот же экземпляр, а затем добавьте столько экземпляров потребителя, сколько требуется для обеспечения нужной пропускной способности:

builder.Services.AddSingleton(_ =>
  Channel.CreateBounded<WorkItem>(
   new BoundedChannelOptions(100)
   {
     FullMode = BoundedChannelFullMode.Wait
   }));
 
// 3 потребителя одного канала
builder.Services.AddHostedService<WorkConsumer>();
builder.Services.AddHostedService<WorkConsumer>();
builder.Services.AddHostedService<WorkConsumer>();

Количество потребителей — регулятор уровня параллелизма. Один потребитель обрабатывает задачи строго последовательно, а 3 — до трёх одновременно. Настраивайте их количество с учётом возможностей последующего этапа обработки: если ProcessAsync обращается к БД, поддерживающей не более 10 одновременных операций записи, не стоит запускать 50 потребителей.

Корректное завершение работы
При остановке приложения в канале могут оставаться необработанные элементы. Если просто завершить процесс, они будут потеряны. Решение – закрыть канал записи при остановке и позволить потребителям обработать оставшиеся данные. За обработку сигнала отвечает небольшой фоновый сервис:

public class ChannelCompleter : IHostedService
{
  private readonly ChannelWriter<WorkItem> _writer;

  public ChannelCompleter(Channel<WorkItem> ch)
    => _writer = ch.Writer;

  public Task StartAsync(CancellationToken ct)
   => Task.CompletedTask;
 
  public Task StopAsync(CancellationToken ct)
  {
    _writer.Complete();
    return Task.CompletedTask;
  }
}

После вызова Complete() метод WriteAsync будет генерировать исключение при попытке записи новым производителем, а метод ReadAllAsync каждого потребителя продолжит выдавать элементы до тех пор, пока буфер не опустеет, после чего цикл завершится. Предоставьте хосту достаточно времени для завершения обработки всех данных, настроив параметр ShutdownTimeout:

builder.Services.Configure<HostOptions>(o =>
  o.ShutdownTimeout = TimeSpan.FromSeconds(30));

Теперь при развёртывании новой версии системы очередь корректно опустошается, а не просто теряет все задачи, находившиеся в процессе обработки.

Когда использовать
Жизненный цикл каналов неразрывно связан с процессом приложения. Если приложение перезапускается, накопленные в буфере элементы исчезают: здесь нет ни записи на диск, ни механизма повторного воспроизведения, ни подтверждения доставки. Это вполне допустимо для задач, потерю которых можно себе позволить или которые можно выполнить заново: например, создание миниатюр изображений, предварительный прогрев кэша или отправка некритичных уведомлений. Однако такой подход неприемлем для обработки платежей или отправки email.
Если вам нужны гарантии сохранности данных при перезапусках, механизмы повторных попыток с обработкой «мертвых» сообщений или распределение задач между разными сервисами — значит, вы переросли возможности каналов. Тогда лучше использовать полноценный брокер сообщений или паттерн Outbox.
Важно осознавать ограничения системы до того, как вы выпустите её в прод, а не после того, как первая же перезагрузка приведет к потере всей очереди.

FAQ
1. В чём разница между ограниченным (bounded) и неограниченным (unbounded) каналом?
Неограниченный канал принимает данные для записи бесконечно, поэтому при быстром производителе и медленном потребителе очередь будет расти, пока не закончится память. Ограниченный канал имеет фиксированную ёмкость; когда он заполнен, операции записи приостанавливаются (или данные отбрасываются, в зависимости от режима обработки переполнения), пока потребитель не обработает накопленные элементы.

2. Чем Channel отличается от BlockingCollection?
BlockingCollection блокирует вызывающий поток, когда коллекция заполнена или пуста, тем самым занимая поток из пула на время ожидания. В канале методы WriteAsync и ReadAsync освобождают поток, поэтому ожидающий производитель или потребитель не потребляет ресурсы потока.

Итого
С появлением System.Threading.Channels реализация паттерна «производитель-потребитель» не требует сторонних библиотек. Ограниченный канал обеспечивает потокобезопасную асинхронную передачу данных с реальным механизмом обратного давления.
Критически важно:
- ограничить размер канала, чтобы всплеск нагрузки не привёл к падению процесса,
- корректно завершать работу писателя при выключении, чтобы накопленные данные обрабатывались, а не отбрасывались.
Если всё сделать правильно, то конечная точка, которая раньше «падала» под наплывом запросов на загрузку, продолжит стабильно работать.

Источник: https://thecodeman.net/posts/producer-consumer-with-channels-in-dotnet