Controlled ThreadPool using .NET Framework 4.0

Introduction

Controlled ThreadPool using BlockingCollection in .NET Framework 4.0.

Background

A very common requirement is to create a pool of defined number of threads and execute tasks on it. For years, We have been controlling the tasks queue collection using locks and having all threads running all the time even when there are no tasks in queue. Another way is to use ManualResetEvents as explained here nicely: http://blogs.msdn.com/b/pfxteam/archive/2010/04/13/9995622.aspx.

Recently I came across the new Threading class (BlockingCollection) available in .NET Framework 4.0 and found that now we can do the same job without locks and without running threads all the time. Also it require less line of neat code.

Using the code

This class is written using generics which means it can store any type of task and one handler for those tasks. It can also be used as a collection for multiple handlers.

public class OrderlyThreadPool<t> : IDisposable
{
    BlockingCollection<t> _workingqueue = null;
    Action<t> _handler = null;
    public OrderlyThreadPool(int wrkerCount, Action<t> handler)
    {
        _workingqueue = new BlockingCollection<t>();
        _handler = handler;
        Worker worker = new Worker(wrkerCount, Process);
        worker.Start();
    }

    public void AddItem(T item)
    {
        _workingqueue.Add(item);
    }
    private void Process()
    {
        foreach (T item in _workingqueue.GetConsumingEnumerable())
        {
            _handler(item);
        }
    }
    public void Dispose() { _workingqueue.CompleteAdding(); 
                            _workingqueue = null; }

}
/*
 * Class which handles the workers on the queue 
 */
public class Worker
{
    int _wrkerCount = 0;
    Action _action = null;
    public Worker(int workerCount, Action action)
    {
        _wrkerCount = workerCount;
        _action = action;

    }
    public void Start()
    {
        // Create and start a separate Task for each consumer:
        for (int i = 0; i < _wrkerCount; i++)
        {
            Task.Factory.StartNew(_action);
        }
    }
}

Caller:

static void main()
{
   OrderlyThreadPool<city> threadpool = new OrderlyThreadPool<city>(6, new Action<city>(Handler));
                threadpool.AddItem(new City() { CityZipCode = "60661",CityName ="Chicago",State="Illinois" });
}
static void Handler(City c)
{
   Console.WriteLine(c.CityName);	
   Console.WriteLine(c.CityZipCode);
   Console.WriteLine(c.State);
}

class City
{
    public string CityName;
    public string CityZipCode;
    public string State;
}
</city></city></city>

Note: This class requires reference to System.Collections.Concurrent,System.Threading.Tasks and System.Collections.Generic
namespaces.

Points of Interest

Threading has been improved significantly with many new classes like LocalThread.