12

I've been doing some work lately with the Reactive Framework and have been absolutely loving it so far. I'm looking at replacing a traditional polling message queue with some filtered IObservables to clean up my server operations. In the old way, I dealt with messages coming into the server like so:

// Start spinning the process message loop
   Task.Factory.StartNew(() =>
   {
       while (true)
       {
           Command command = m_CommandQueue.Take();
           ProcessMessage(command);
       }
   }, TaskCreationOptions.LongRunning);

Which results in a continuously polling thread that delegates commands from clients out to the ProcessMessage method where I have a series of if/else-if statements that determine the type of the command and delegate work based on its type

I am replacing this with an event driven system using Reactive for which I've written the following code:

 private BlockingCollection<BesiegedMessage> m_MessageQueue = new BlockingCollection<BesiegedMessage>();
 private IObservable<BesiegedMessage> m_MessagePublisher;

 m_MessagePublisher = m_MessageQueue
       .GetConsumingEnumerable()
       .ToObservable(TaskPoolScheduler.Default);

        // All generic Server messages (containing no properties) will be processed here
 IDisposable genericServerMessageSubscriber = m_MessagePublisher
       .Where(message => message is GenericServerMessage)
       .Subscribe(message =>
       {
           // do something with the generic server message here
       }

My question is that while this works, is it good practice to use a blocking collection as the backing for an IObservable like this? I don't see where Take() is ever called this way which makes me think that the Messages will pile off on the queue without being removed after they have been processed?

Would it be more efficient to look into Subjects as the backing collection to drive the filtered IObservables that will be picking up these messages? Is there anything else I'm missing here that might benefit the architecture of this system?

Improve this question
5
  • 1
    +1 for using Rx :) - what's your command queue - messages in its nature? Is it of temp character - or something available app-wide - so one part submits - other part subscribes? Or i.e. if two subscribers 'subscribe' (what else) do they get to see the same things as they happen, or each has its own ? - when dealing w/ Rx you need to answer those simple things - the nature, use. Just throwing it here fast if it helps Commented Apr 2, 2013 at 20:55
  • 1
    @NSGaga The messages are underlying to the entire application as there is two way communication happening here both to the server from the client and vice versa. The messages are simply a common language that both can act on Commented Apr 2, 2013 at 21:27
  • First thing that pops into my head is that you don't particularly need a BlockingCollection any longer - you need those for when you are actively polling for new messages, but if you are having your messages delivered to you, there's no need to "block until I get a signal" any longer Commented Apr 2, 2013 at 22:02
  • 1
    @JerKimball That thought occurred to me as well. In my system, all messages are sent to the server through a client exposed method called SendMessage(string message). Is there any way to make an IObservable source from all messages that get delivered here as opposed to putting them into the BlockingCollection? Commented Apr 2, 2013 at 22:23
  • For the record, the messages definitely get taken off the BlockingCollection queue as they are processed. Thanks again for the superb example code!! Commented Apr 26, 2014 at 22:24

4 Answers 4

Reset to default
8

Here is a complete worked example, tested under Visual Studio 2012.

  1. Create a new C# console app.
  2. Right click on your project, select "Manage NuGet Packages", and add "Reactive Extensions - Main Library".

Add this C# code:

using System;
using System.Collections.Concurrent;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
namespace DemoRX
{
    class Program
    {
        static void Main(string[] args)
        {
            BlockingCollection<string> myQueue = new BlockingCollection<string>();
            {                
                IObservable<string> ob = myQueue.
                  GetConsumingEnumerable().
                  ToObservable(TaskPoolScheduler.Default);

                ob.Subscribe(p =>
                {
                    // This handler will get called whenever 
                    // anything appears on myQueue in the future.
                    Console.Write("Consuming: {0}\n",p);                    
                });
            }
            // Now, adding items to myQueue will trigger the item to be consumed
            // in the predefined handler.
            myQueue.Add("a");
            myQueue.Add("b");
            myQueue.Add("c");           
            Console.Write("[any key to exit]\n");
            Console.ReadKey();
        }
    }
}

You will see this on the console:

[any key to exit]
Consuming: a
Consuming: b
Consuming: c

The really nice thing about using RX is that you can use the full power of LINQ to filter out any unwanted messages. For example, add a .Where clause to filter by "a", and observe what happens:

ob.Where(o => (o == "a")).Subscribe(p =>
{
    // This will get called whenever something appears on myQueue.
    Console.Write("Consuming: {0}\n",p);                    
});

Philosophical notes

The advantage of this method over starting up a dedicated thread to poll the queue, is that you don't have to worry about disposing of the thread properly once the program has exited. This means you don't have to bother with IDisposable or CancellationToken (which is always required when dealing with a BlockingCollection or else your program might hang on exit with a thread that refuses to die).

Believe me, its not as easy as you think to write completely robust code to consume events coming out of a BlockingCollection. I much prefer using the RX method, as shown above as its cleaner, more robust, has less code, and you can filter using LINQ.

Latency

I was surprised at how fast this method is.

On my Xeon X5650 @ 2.67Ghz, it takes 5 seconds to process 10 million events, which works out at approximately 0.5 microseconds per event. It took 4.5 seconds to put the items into the BlockingCollection, so RX was taking them out and processing them almost as fast as they were going in.

Threading

In all of my tests, RX only spun up one thread to handle the tasks on the queue.

This means that we have a very nice pattern: we can use RX to collect incoming data from multiple threads, place them into a shared queue, then process the queue contents on a single thread (which is, by definition, thread safe).

This pattern eliminates a huge amount of headaches when dealing with multithreaded code, by decoupling the producer and consumer of data via a queue, where the producer could be multi-threaded and the consumer is single-threaded and thus thread-safe. This is the concept that makes Erlang so robust. For more information on this pattern, see Multi-threading made ridiculously simple.

Improve this answer
Sign up to request clarification or add additional context in comments.

4 Comments

Really not sure what the purpose of this answer is. The question is extremely old not to mention you simply repeated the code snippet I had posted. It adds nothing constructive.
Its more for my own reference as the example you initially gave was incomplete. I'm following the same learning curve that you followed a year ago.
Agree with Jesse. It is nice what you have done here, but I think you could replace all of that with a subject and .Synchonize() to allow multiple threads calling OnNext(). That was my point in my answer, maybe it was not well articulated. Basically a queue requires a thread to be dedicated to reading from it. A background thread will tear down when the process exits. Also Linq works great on IEnumerable<T>.
On pondering this, the indirection of a blocking Queue does separate the slowness of the Subscriber's OnNext handler from blocking the producers ability to en-queue things. However again, this is easily solved with an ObserveOn(IScheduler) which also re-introduces the same concurrency as the answer provided above. So the queue.ToObservable code could also just be subject.Synchronize().ObserveOn(TaskPoolScheduler.Default), no? An potentially remove the Synchronize() as the ObserveOn will serialize the data with it's internal queue. So I think subject.ObserveOn(TaskPoolScheduler.Default)
6

Here's something pulled directly from my posterior - any real solution would be very much dependent on your actual usage, but here's "The cheapest pseudo Message Queue system ever":

Thoughts/motivations:

  • Deliberate exposure of IObservable<T> such that subscribers can do any filtering/cross subscriptions they want to
  • The overall Queue is typeless, but Register and Publish are type-safe(ish)
  • YMMV with the Publish() where it is - try experimenting with moving it around
  • Generally Subject is a no-no, although in this case it does make for some SIMPLE code.
  • One could "internalize" the registration to actually do the subscription as well, but then the queue would need to manage the IDisposables created - bah, let your consumers deal with it!

The Code:

public class TheCheapestPubSubEver
{    
    private Subject<object> _inner = new Subject<object>();

    public IObservable<T> Register<T>()
    {
        return _inner.OfType<T>().Publish().RefCount();
    }
    public void Publish<T>(T message)
    {
        _inner.OnNext(message);
    }
}

Usage:

void Main()
{
    var queue = new TheCheapestPubSubEver();

    var ofString = queue.Register<string>();
    var ofInt = queue.Register<int>();

    using(ofInt.Subscribe(i => Console.WriteLine("An int! {0}", i)))
    using(ofString.Subscribe(s => Console.WriteLine("A string! {0}", s)))
    {
        queue.Publish("Foo");
        queue.Publish(1);
        Console.ReadLine();
    }
}

Output:

A string! Foo
An int! 1

HOWEVER, this doesn't strictly enforce "consuming consumers" - multiple Registers of a specific type would result in multiple observer calls - that is:

var queue = new TheCheapestPubSubEver();

var ofString = queue.Register<string>();
var anotherOfString = queue.Register<string>();
var ofInt = queue.Register<int>();

using(ofInt.Subscribe(i => Console.WriteLine("An int! {0}", i)))
using(ofString.Subscribe(s => Console.WriteLine("A string! {0}", s)))
using(anotherOfString.Subscribe(s => Console.WriteLine("Another string! {0}", s)))

{
    queue.Publish("Foo");
    queue.Publish(1);
    Console.ReadLine();
}

Results in:

A string! Foo
Another string! Foo
An int! 1
Improve this answer

1 Comment

Yes - multiple Publish-es - means multiple feeds. You can though make all that (and string/int combined) into a single Publish. Just for the OP-s clarity.
1

I haven't used BlockingCollection in this context - so I'm 'conjecturing' - you should run it to approve, disprove.

BlockingCollection might only further complicate things here (or provide little help). Take a look at this post from Jon - simply to confirm. GetConsumingEnumerable will provide you with 'per subscriber' enumerable. Exhausting them down eventually - something to have in mind with Rx.

Also the the IEnumerable<>.ToObservable further flattens out the 'source'. As it works (you can lookup the source - I'd recommend w/ Rx more than anything) - each subscribe creates an own 'enumerator' - so all will be getting their own versions of the feed. I'm really not sure, how that pans out in the Observable scenario like this.

Anyhow - if you want to provide app-wide messages - IMO you'd need to introduce Subject or state in some other form (e.g. Publish etc.). And in that sense, I don't think BlockingCollection will help any - but again, it's best that you try it out yourself.

Note (a philosophical one)

If you want to combine message types, or combine different sources - e.g. in a more 'real world' scenario - it gets more complex. And it gets quite interesting I must say.

Keep an eye on having them 'rooted' into a single-shared stream (and avoid what Jer suggested rightly).

I'd recommend that you don't try to evade using Subject. For what you need, that's your friend - no matter all the no-state related discussions and how Subject is bad - you effectively have a state (and you need a 'state') - Rx kicks in 'after the fact', so you enjoy benefits from it regardless.

I encourage you to go that way, as I love it how it turned out.

Improve this answer

Comments

-1

My issue here is that we have turned a Queue (which I normally associate with destructive reads by one consumer especially if you are using BlockingCollection) into a broadcast (send to anyone and everyone listening right now).

These seem two conflicting ideas.

I have seen this done, but it then was thrown away as it was the "right solution to the wrong question".

Improve this answer

2 Comments

In the example code above, you simply cannot add multiple subscriptions to a single BlockingCollection, as each subscriber will do a destructive read on the BlockingCollection. The example code above is simply a nice way to cleanly process items that are added to a queue at any point in the future.
And how is that different from not using Rx and just having a thread dedicated to consuming from the Queue? While I love Rx, I dont think it pays for itself in this case.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.