C# inter-thread communication

Question

I want two threads to collaborate; a producer and a consumer. the consumer is rather slow, and the producer is very fast and works in bursts.

for example the consumer can process one message per 20 seconds, and the producer can produce 10 messages in one second, but does it about once in a long while so the consumer can catch up.

I want something like:

Stream commonStream;
AutoResetEvent commonLock;

void Producer()
{
    while (true)
    {
        magic.BlockUntilMagicAvalible();
        byte[] buffer = magic.Produce();
        commonStream.Write(buffer);
        commonLock.Set();
    }
}

void Consumer()
{
    while(true)
    { 
        commonLock.WaitOne();
        MagicalObject o = binarySerializer.Deserialize(commonStream);
        DoSomething(o);
    }
}

Answer 1

If you have .Net 4.0 or higher you can do it this way by using a BlockingCollection

int maxBufferCap = 500;
BlockingCollection<MagicalObject> Collection 
                           = new BlockingCollection<MagicalObject>(maxBufferCap);
void Producer()
{
    while (magic.HasMoreMagic)
    {
        this.Collection.Add(magic.ProduceMagic());
    }
    this.Collection.CompleteAdding();
}

void Consumer()
{
    foreach (MagicalObject magicalObject in this.Collection.GetConsumingEnumerable())
    {
        DoSomthing(magicalObject);
    }
}

The foreach line will sleep if there is no data in the buffer, it will also automatically wake it self up when something is added to the collection.

The reason I set the max buffer is if your producer is much faster than the consumer you may end up consuming a lot of memory as more and more objects get put in to the collection. By setting up a max buffer size as you create the blocking collection when the buffer size is reached the Add call on the producer will block until a item has been removed from the collection by the consumer.

Another bonus of the BlockingCollection class is it can have as many producers and consumers as you want, it does not need to be a 1:1 ratio. If DoSomthing supports it you could have a foreach loop per core of the computer (or even use Parallel.ForEach and use the consuming enumerable as the data source)

void ConsumersInParalell()
{
    //This assumes the method signature of DoSomthing is one of the following:
    //    Action<MagicalObject>
    //    Action<MagicalObject, ParallelLoopState>
    //    Action<MagicalObject, ParallelLoopState, long>
    Paralell.ForEach(this.Collection.GetConsumingEnumerable(), DoSomthing);
}