使用BlockingCollection将Reactive Framework作为消息队列

Question

我最近在Reactive Framework上做了一些工作,到目前为止我一直非常喜欢它.我正在寻找用一些过滤的IObservable替换传统的轮询消息队列来清理我的服务器操作.以旧的方式,我处理进入服务器的消息,如下所示:

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

这导致连续轮询线程将命令从客户端委托给ProcessMessage方法,其中我有一系列if/else-if语句,用于确定命令的类型并根据其类型委派工作

我用一个使用Reactive的事件驱动系统替换它,我已经编写了以下代码:

 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
       }

我的问题是,虽然这有效,但是使用阻塞集合作为这样的IObservable的支持是一种好习惯吗？我没有看到Take()被称为这样的方式,这让我觉得消息将在队列中堆积而不会在处理后被删除？

将Subjects作为支持集合来驱动过滤后的IObservables来获取这些消息会更有效吗？还有什么我在这里缺少可能有益于这个系统架构的东西吗？

Answer 1

这是一个完整的工作示例,在Visual Studio 2012下进行了测试.

1. 创建一个新的C#控制台应用程序.
2. 右键单击您的项目,选择"Manage NuGet Packages",然后添加"Reactive Extensions - Main Library".

添加此C#代码:

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();
        }
    }
}

您将在控制台上看到:

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

使用RX非常好的一点是,您可以使用LINQ的全部功能来过滤掉任何不需要的消息.例如,添加一个.Where子句以"a"过滤,并观察发生的情况:

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

哲学笔记

这种方法优于启动专用线程来轮询队列的优点是,一旦程序退出,您不必担心正确处理线程.这意味着您不必费心使用IDisposable或CancellationToken(在处理BlockingCollection时总是需要它,否则您的程序可能会在退出时挂起并拒绝死亡).

相信我,它并不像你想象的那样容易编写完全健壮的代码来消耗来自BlockingCollection的事件.我更喜欢使用RX方法,如上所示,它更干净,更健壮,代码更少,您可以使用LINQ进行过滤.

潜伏

我很惊讶这种方法的速度有多快.

在我的Xeon X5650 @ 2.67Ghz上,处理1000万个事件需要5秒钟,每个事件大约0.5微秒.将这些项目放入BlockingCollection需要4.5秒,因此RX将它们取出并按照它们进入的速度处理它们.

穿线

在我的所有测试中,RX只会旋转一个线程来处理队列中的任务.

这意味着我们有一个非常好的模式:我们可以使用RX从多个线程收集传入数据,将它们放入共享队列,然后在单个线程(根据定义,线程安全)处理队列内容.

这种模式在处理多线程代码时消除了大量的麻烦,通过队列解耦数据的生产者和消费者,生产者可以是多线程的,而消费者是单线程的,因此是线程安全的.这是使Erlang如此强大的概念.有关此模式的更多信息,请参阅多线程变得非常简单.

Answer 2

这里的东西直接从我的后拉-任何真正的解决办法是非常多的依赖于你的实际使用情况,但这里的"最便宜的假消息队列系统不断":

思想/动机:

• 有意识地接触IObservable<T>订户可以进行他们想要的任何过滤/交叉订阅
• 整体队列是无类型,但Register和Publish是类型安全(ISH)
• YMMV与Publish()它的位置 - 尝试移动它
• 通常Subject是禁止使用,虽然在这种情况下它确实产生了一些SIMPLE代码.
• 人们可以"内化"注册以实际进行订阅,但是队列需要管理IDisposables创建的 - bah,让您的消费者处理它!

代码:

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);
    }
}

用法:

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();
    }
}

输出:

A string! Foo
An int! 1

但是,这并没有严格执行"消费者" - 特定类型的多个注册会导致多个观察者调用 - 即:

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();
}

结果是:

A string! Foo
Another string! Foo
An int! 1