Как реализовать непрерывно выполняемые блоки потока данных в TPL?

Question

У меня есть настройка блока потока данных производителя/потребителя с использованием BufferBlock и ActionBlock, и он отлично работает в консольном приложении;

После добавления всех предметов в BurfferBlock и связывания BufferBlock с другими элементами действия; он работает хорошо.

Теперь я хочу использовать эту внутреннюю службу, где этот конвейер блока потока данных всегда будет вверх, и когда сообщения будут доступны через внешние события, он войдет в буферный блок и начнет обработку. Как я могу добиться этого?

До сих пор я сделал ниже:

public void SetupPipeline() 
{ 

     FirstBlock = new ActionBlock<WorkItem>(new Action<WorkItem>(ProcessIncomingMessage), 
     new ExecutionDataflowBlockOptions 
     { 
      MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded 
     }); 

     BufferBlock = new BufferBlock<WorkItem>(); 

     GroupingDataflowBlockOptions GroupingDataflowBlockOptions = new GroupingDataflowBlockOptions(); 
     GroupingDataflowBlockOptions.Greedy = true; 
     GroupingDataflowBlockOptions.BoundedCapacity = GroupingDataflowBlockOptions.Unbounded; 
     CancellationTokenSource = new CancellationTokenSource(); 
     CancellationToken = CancellationTokenSource.Token; 
     GroupingDataflowBlockOptions.CancellationToken = CancellationToken; 
     BatchBlock = new BatchBlock<WorkItem>(BoundingCapacity, GroupingDataflowBlockOptions); 

     ProcessItems = new ActionBlock<WorkItem[]>(WorkItems => 
      ProcessWorkItems(WorkItems.ToList<WorkItem>()), 
      new ExecutionDataflowBlockOptions 
      { 
       CancellationToken = CancellationToken 
      }); 

     Timer = new Timer(_ => 
       BatchBlock.TriggerBatch() 
      ); 

     TimingBlock = new TransformBlock<WorkItem, WorkItem>(WorkItem => 
     { 
      Timer.Change(TimerInterval, Timeout.Infinite); 
      logger.Debug("Inside TimingBlock : " + WorkItem.ToString()); 
      return WorkItem; 
     }, new ExecutionDataflowBlockOptions 
     { 
      CancellationToken = CancellationToken 
     }); 

     BatchBlock.LinkTo(ProcessItems); 
     TimingBlock.LinkTo(BatchBlock); 
     BufferBlock.LinkTo(TimingBlock); 
} 

Answer 1

Размер партии определяется переменной «BoundingCapacity» в конструкторе batchblock. Партия будет размещена в следующих случаях:

• ряд сообщений, равных размера партии были получены (определены конструктором)
• Пакетный блок помечается для завершения
• Метод triggerbatch называется

Кажется, что вы хотите, чтобы партия отправлялась после того, как был достигнут размер ванны или произошел тайм-аут. Если это так, и если размер партии не является критическим, я бы просто добавил повторяющийся интервал к таймеру, который у вас есть, и сделайте объект, расположенный ниже по потоку от блока batchblock, игнорировать пустые записи.

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

Проблема с попыткой реализовать таймер тайм-аута вокруг пакетного блока, который срабатывает только с первого триггера, заключается в том, что вам нужно будет подсчитывать и проверять сообщения в буферном блоке, или вам нужно будет просматривать сообщения из буфера. Оба этих сценария создадут много уродства и/или нарушают инкапсуляцию блоков.

Answer 2

В упрощении, DataFlow является способом обработки кучи объектов, используя набор методов. Он не предусматривает и не ожидает какого-либо конкретного способа создания этих объектов.

Если вы хотите, чтобы трубопровод оставался в живых, просто не завершайте приложение. Если вы не хотите использовать консольное приложение, создайте службу, которая строит конвейер и отправляет ему объекты до его закрытия.

Сообщения - это просто объекты, которые вы создадите, читая данные, в ответ на события (независимо от того, что это означает) или любым другим способом.

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

• Если данные поступают из другого консольного приложения, вы можете передать результаты одного приложения к другому, анализировать данные, поступающие из входного потока приложения командной строки, создать сообщений и передать их в конвейер
• Если вы хотите прослушивать запросы для служб, вы можете разместить службу .NET Pipe, WCF или Web API для прослушивания вызовов и передачи опубликованных данных в конвейер.
• Если данные поступают из базы данных, вы можете опросить изменения и отправить любые измененные данные в конвейер.

Дело в том, что Dataflow предназначен для обработки данных, а не для прослушивания событий. Это не полная система распределенных агентов, если это то, что вы искали.