Event-based multithread producer-consumer?

Question

I wish to figure out how to create a multithreaded producer/consumer program where producing is based on an external event.

I have built a synchronized queue having 2 thread loops, one to enqueue and one to dequeue and write into a file, with classic pattern:

Class SyncQueue  
{
    ...
    producerThread = new Thread(new ThreadStart(StartProducer));
    consumerThread = new Thread(new ThreadStart(StartConsumer));
    ...

    public void Start()
    {
        producerThread.Start();
        consumerThread.Start();
    }

    public void StartProducer()
    {
        while (!stop)
        { 
            //wait for an external event to happen 
            //Data data = eventHandler() : How to wait for an event and get?
            Enqueue(data);
        }
    }
}

In the other side, I have a method in another class that handles an external event independantly.

public void OnExternalEvent()
{
    //event occured 
    //how to notify and send data to the producer thread?
}

My question in not about the producer/consumer pattern, but it's about integrating events among it.

The code I posted is just to make my question clearer, I added two concrete questions in comments.

I would appreciate if anyone can tell me how to do it. I am new in C# and do not have a strong background with event handling.

Answer 1

Ditch your custom synchronized queue and use a BlockingCollection< T >.

With BlockingCollection, you don't have separate threads controlling the queue. Rather, you have a queue, and threads can directly enqueue or dequeue items. The data structure itself handles any concurrency issues, and does non-busy waits when trying to dequeue.

See https://stackoverflow.com/a/19848796/56778 and https://stackoverflow.com/a/19823345/56778 for examples, or just search around a bit. Also, see my Simple Multithreading blog post for a little bit more detail.

Answer 2

Just consider using Rx (Reactive extensions) for that. They are modern, flexible, convenient, support multithreading and, simply, just cool. he he

With using IObservable<T> collections you just subscribe to them and receive notifications if something changes (is posted) into them.

This is how you post the data:

var subject = new Subject<string>();
subject.OnNext("Value");
subject.OnCompleted();

This is how subscribing:

subject.SubscribeOn(Scheduler.CurrentThread);  // this is for thread safety, for instance
subject.Subscribe(() => { /* some action */ });

Oh, and, also, all that is thread safe!