поток данных tpl: фиксированный размер буфера без отбрасывания элементов

Question

После игры с потоком данных я столкнулся с новой проблемой. Я хотел бы ограничить входной уровень всех блоков. Мой productionblock (ActionBlock) быстро создает 5000 элементов и помещает их в широковещательный блок. Поэтому, если я установил BoundedCapacity в широковещательном блоке на 100, он выбросит много данных. Но я бы предпочел, чтобы производственный блок дождался новых слотов во входном буфере моего буфера.

Есть ли способ избавиться от этой проблемы?

Answer 1

Это точно, что BufferBlock для. Если вы установите его BoundedCapacity и он будет заполнен, он отложит получение любых сообщений, пока кто-то их не будет использовать. Это означает, что, например, Post() будет заблокирован, а SendAsync() вернет незавершенный Task.

EDIT: Нет встроенного блока, который отправляет несколько целей и никогда не удаляет данные. Но вы можете легко построить один из себя ActionBlock и представляемого цикла:

static ITargetBlock<T> CreateMultipleTargetsBlock<T>(
    IEnumerable<ITargetBlock<T>> targets, int boundedCapacity) 
{ 
    var targetsList = targets.ToList(); 

    var block = new ActionBlock<T>(
     async item => 
     { 
      foreach (var target in targetsList) 
      { 
       await target.SendAsync(item); 
      } 
     }, 
     new ExecutionDataflowBlockOptions { BoundedCapacity = boundedCapacity }); 

    // TODO: propagate completion from block to targets 

    return block; 
} 

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