С# задача многозадачное дросселирование

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

Очередь 1 → 3 потока;
Очередь 2 → 6 потоков;

Вид системы задач. Мне удалось реализовать сам, используя простой старый код С# (он же System.Threading.Thread, lock и queue), который работает более чем на 1 год. Тем не менее, я продолжаю читать статьи о чудесах TaskFactory и TaskScheduler, о том, что это возможно со встроенными классами в .NET, но я не смог найти пример, чтобы доказать это. Я бы хотел протестировать его и сравнить с тем, что у меня есть сейчас, чтобы увидеть, работает ли он лучше, и если это произойдет, чтобы заменить его.

Более того, я могу жить без необходимости ограничивать/устанавливать количество параллельных потоков для каждой очереди, если я могу получить guarantee, если объект, предназначенный для очереди # 2, выполняется imediatly, даже если очередь # 1 выполнение при полной нагрузке.

Итак, мой вопрос: есть ли что-то в .net 4 и более, может кто-нибудь указать мне на образец? Я ищу одну целую неделю и не смог получить что-то важное.

Ответ 1

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

Ответ 2

Это на самом деле довольно тривиально, используя TPL и новые коллекции в System.Collections.Concurrent.

Для ваших нужд BlockingCollection<T> - это то, что я бы рекомендовал. По умолчанию он использует ConcurrentQueue<T> в качестве основного хранилища, который идеально подходит для того, что вы хотите.

var queue = new BlockingCollection<Message>();

Чтобы установить некоторый код, работающий над этими сообщениями, и контролировать, сколько из них может выполняться параллельно, так же просто:

//Set max parallel Tasks
var options = new ParallelOptions
{
    MaxDegreeOfParallelism = 10
};

Parallel.ForEach(queue.GetConsumingEnumerable(), options, msg =>
{
    //Do some stuff with this message
});

Итак, что здесь происходит? Ну...

Вызов GetConsumingEnumerable() будет блокироваться до тех пор, пока не будет что-то в queue. Это здорово, потому что дополнительный код не нужен, чтобы сигнализировать о готовности новой работы. Скорее, когда queue заполняется, новая задача с вашим (анонимным) делегатом будет выведена с помощью элемента.

Объект ParallelOptions позволяет вам контролировать, как Parallel.ForEach. В этом случае вы говорите, что вы никогда не хотите выполнять более 10 заданий в любой момент времени. Важно отметить, что Tasks!= Threads. Детали мрачны, но, разумеется, есть много оптимизаций под капотом. Все это подключается к вам, но это не для слабонервных.

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