Microsoft TPL Dataflow — синхронная обработка коррелирующих запросов

1. Вы говорите, что хотите обрабатывать некоторые запросы параллельно (по крайней мере, я предполагаю, что вы имели в виду «асинхронно»), но ActionBlock по умолчанию не является параллельным. Чтобы изменить это, установите MaxDegreeOfParallelism .

2. Вы пытаетесь использовать TryAdd() в качестве фильтра, но это не сработает по двум причинам:

   1. The filter is invoked only once, it's not automatically retried or anything like that. That means that if an item doesn't go through, it would never go through, even after the item that was blocking it was completed.
   2. Если элемент застрял в очереди вывода блока, никакие другие элементы не выберутся из этого блока. Это может значительно снизить уровень параллелизма, даже если вы каким-то образом обошли предыдущую проблему.

3. Я думаю, что самым простым решением здесь было бы иметь блок для каждой группы, таким образом элементы из каждой группы будут обрабатываться последовательно, а элементы из разных групп будут обрабатываться параллельно. В коде это может выглядеть примерно так:

   var processingBlocks = new Dictionary<string, ActionBlock<Request>>();
   
   var splitterBlock = new ActionBlock<Request>(request =>
   {
       ActionBlock<Request> processingBlock;
   
       if (!processingBlocks.TryGetValue(request.Id, out processingBlock))
       {
           processingBlock = processingBlocks[request.Id] =
               new ActionBlock<Request>(r => /* process the request here */);
       }
   
       processingBlock.Post(request);
   });
   

   Проблема с этим подходом заключается в том, что блоки обработки для групп никогда не исчезают. Если вы не можете себе это позволить (это утечка памяти), потому что у вас будет большое количество групп, тогда хеширование подход, предложенный И3арноном, является правильным.

Я полагаю, это потому, что ваш LinkTo() настроен неправильно. Имея LinkTo() и передавая функцию в качестве аргумента, вы добавляете условие. Итак, эта строка:

bufferBlock.LinkTo(actionBlock, x => requestTracker.TryAdd(x.Id, x.Name));

По сути, речь идет о передаче данных из bufferBlock в actionBlock, ЕСЛИ вы можете добавить в свой параллельный словарь, что не обязательно имеет смысл (по крайней мере, в вашем примере кода)

Вместо этого вы должны связать свой bufferBlock с блоком действий без лямбды, поскольку в этой ситуации вам не нужна условная ссылка (по крайней мере, я так не думаю, основываясь на вашем примере кода).

Кроме того, взгляните на этот вопрос SO, чтобы узнать, следует ли вам использовать SendAsync() или Post(), поскольку Post() проще обрабатывать для простого добавления данных в конвейер: TPL Dataflow, в чем функциональное различие между Post() и SendAsync()? . SendAsync вернет задачу, а Post вернет значение true/false в зависимости от успешного входа в конвейер.

Таким образом, чтобы выяснить, что происходит не так, вам нужно обработать продолжения ваших блоков. В MSDN есть хороший учебник по их введению в TPL Dataflow: Создайте конвейер потока данных По сути это будет выглядеть так:

//link to section
bufferBlock.LinkTo(actionBlock);
//continuations
bufferBlock.Completion.ContinueWith(t =>
{
     if(t.IsFaulted)  ((IDataFlowBlock).actionBlock).Fault(t.Exception); //send the exception down the pipeline
     else actionBlock.Complete(); //tell the next block that we're done with the bufferblock
 });

Затем вы можете поймать исключение (AggregateException) при ожидании конвейера. Вам действительно нужно использовать concurrentdictionary в вашем фактическом коде для отслеживания, потому что это может вызвать проблему, когда он не может добавить, поскольку, когда предикат linkto возвращает false, он не передает данные в следующий блок конвейера .