gpt4 book ai didi

c# - 使用 BlockingCollection 作为消息队列的响应式(Reactive)框架

转载 作者:可可西里 更新时间:2023-11-01 08:08:20 25 4
gpt4 key购买 nike

我最近一直在使用 Reactive Framework 做一些工作,到目前为止我非常喜欢它。我正在考虑用一些过滤的 IObservables 替换传统的轮询消息队列来清理我的服务器操作。在过去,我处理进入服务器的消息是这样的:

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

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

我将其替换为使用 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 会更有效吗?还有什么我在这里遗漏的可能有利于这个系统的架构吗?

最佳答案

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

  1. 创建一个新的 C# 控制台应用。
  2. 右键单击您的项目,选择“管理 NuGet 包”,然后添加“ react 性扩展 - 主图书馆”。

添加此 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 如此健壮的概念。有关此模式的更多信息,请参阅 Multi-threading made ridiculously simple .

关于c# - 使用 BlockingCollection 作为消息队列的响应式(Reactive)框架,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/15773008/

25 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com