Multiple producers, single consumer

I have to develop a multithreaded application, where there will be multiple threads, each thread generates custom event log which need to be saved in queue (not Microsoft MSMQ).

There will be another thr开发者_开发知识库ead which reads log data from queue and manipulates it, with certain information to save log information into a file. Basically here we are implementing Multiple-producer, Single-consumer paradigm.

Can anybody provide suggestions on how to implement this in C++ or C#.

Thanks,

This kind of thing is very easy to do using the BlockingCollection<T> defined in System.Collections.Concurrent.

Basically, you create your queue so that all threads can access it:

BlockingCollection<LogRecord> LogQueue = new BlockingCollection<LogRecord>();

Each producer adds items to the queue:

while (!Shutdown)
{
    LogRecord rec = CreateLogRecord(); // however that's done
    LogQueue.Add(rec);
}

And the consumer does something similar:

while (!Shutdown)
{
    LogRecord rec = LogQueue.Take();
    // process the record
}

By default, BlockingCollection uses a ConcurrentQueue<T> as the backing store. The ConcurrentQueue takes care of thread synchronization and, and the BlockingCollection does a non-busy wait when trying to take an item. That is, if the consumer calls Take when there are no items in the queue, it does a non-busy wait (no sleeping/spinning) until an item is available.

You can use a synchronized queue (if you have .NET 3.5 or older code) or even better the new ConcurrentQueue<T>!

What you are planning is a classic producer consumer queue with a thread consuming the items on the queue to do some work. This can be wrapped into is a higher level construct called an "actor" or "active object".

Basically this wraps the queue and the thread that consumes the items into a single class, the other threads all asynchronous methods on this class with put the messages on the queue to be performed by the actor's thread. In your case the class could have a single method writeData which stores the data in the queue and triggers the condition variable to notify the actor thread that there is something in the queue. The actor thread sees if there is any data in the queue if not waits on the condition variable.

Here is a good article on the concept:

http://www.drdobbs.com/go-parallel/article/showArticle.jhtml;jsessionid=UTEXJOTLP0YDNQE1GHPSKH4ATMY32JVN?articleID=225700095