Alex Rivera | Logout

c# task multi-queue throttling

Asked 2012-03-25T01:44:27.510
12

I need a environment which needs to maintain different task queues, and for each of them to have a well defined number of concurrent threads that can execute for each queue. Something like this:

  • Queue 1 -> 3 threads;
  • Queue 2 -> 6 threads;

Kind of Task system. I have managed to implement by myself this using plain old c# code (aka System.Threading.Thread, lock and queue) which works more than fine for 1+ year. However, I keep reading articles about the wonders of TaskFactory and TaskScheduler, about being possible this things with built-in classes in .NET, but I have failed to find an example to prove this. I would like to test it and to compare with what I have right now to see if it's working better and if it does, to replace it.

More, I can live without having to limit/set the number of parallel threads for each queue as long as I can get the guarantee that if an item targeted for queue #2 is executed imediatly even if queue #1 is executing on full load.

So, my question is - is there something in .net 4 and more, can someone point me to a sample? I am looking for one an entire week and failed to get something relevant.

c# queue task-parallel-library task
Edit
Report

1 Answer

8

This is actually pretty trivial using the TPL and the new collections in System.Collections.Concurrent.

For your needs the BlockingCollection<T> is what I would recommend. By default it uses a ConcurrentQueue<T> as the underlying store which is perfect for what you want.

var queue = new BlockingCollection<Message>();

To set some code working on those messages, and control how many can execute in parallel is as simple as this:

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

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

So what is going on here? Well...

The call to GetConsumingEnumerable() will actually block until there is something in queue to consume. This is great because no extra code is necessary for signaling that new work is ready to be done. Rather, as queue fills up, a new Task with your (anonymous) delegate will be kicked off with an item.

The ParallelOptions object allows you to control how Parallel.ForEach operates. In this case, you are telling it you neve

answered 2012-03-25T02:34:17.590

Your Answer