Дождитесь блокировки коллекции (очереди) для уменьшения размера в C#

Question

Я работаю над проектом со следующим рабочим процессом:

Часть первая:

• Событие прибывает асинхронно и очередями в блокировании очередь, мы будем называть это Q1
• Thread подхватывает следующий доступный элемент из этой очереди
• Пункт заканчивается работает {N} ряд задач параллельно
• Каждое задание ставит в очередь его результат во второй очереди, мы будем называть это Q2.
• При обработке финишных элементов следующий элемент считывается из очереди.

Часть вторая:

• Другой поток считывает из Q2 один объект в то время, и работает на результат

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

---

Мой вопрос

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

Answer 1

Вместо использования Queue<T> вы можете использовать BlockingCollection<T> для Q2. Если вы установите BoundedCapacity, вызовы Q2.Add() будут блокироваться при достижении емкости. Это автоматически закроет обработку Q1, так как задачи N начнут блокироваться, если они не смогут добавить в конечную очередь.

Answer 2

Я предполагаю, что вы получаете данные в случайных наводнениях, с длинными засухами, во время которых Q2 может догнать. Рассматривали ли вы просто ограничение количества параллельных потоков, порожденных из Q1, используя ограниченный пул потоков для этих задач?

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

Answer 3

Ваша проблема кажется прекрасным примером для решения библиотеки TPL Dataflow.Если вы готовы попробовать, вот как она могла бы работать (это очень простой пример, конечно):

TransformBlock<int, bool> transform = new TransformBlock<int, bool>(i => i > 5 ? true : false, 
      new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 }); 
ActionBlock<bool> resultBlock = new ActionBlock<bool>(b => Console.WriteLine("My result is : " + b), 
      new ExecutionDataflowBlockOptions { BoundedCapacity = 10 }); 
transform.LinkTo(resultBlock); 

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

Затем вы создаете второй блок (работающий как ваш Q2), который будет иметь BoundedCapacity, и он обрабатывает каждое сообщение синхронно, вызывая действие для каждого элемента. Этот блок может быть заменен любым другим, например BufferBlock, который позволит вам опросить его по запросу.