c# - 为什么不建议在 .NET Reactive Extensions 中使用 Subjects？

Question

我目前正在掌握 .NET 的 Reactive Extensions 框架，并且正在研究我发现的各种介绍资源(主要是 http://www.introtorx.com)

我们的应用程序涉及许多检测网络帧的硬件接口(interface)，这些将是我的 IObservables，然后我有各种组件将使用这些帧或对数据执行某种方式的转换并生成新类型的帧.例如，还会有其他组件需要显示每第 n 帧。我相信 Rx 将对我们的应用程序有用，但是我正在为 IObserver 接口(interface)的实现细节而苦苦挣扎。

我阅读的大部分(如果不是全部)资源都说我不应该自己实现 IObservable 接口(interface)，而应该使用提供的函数或类之一。从我的研究看来，创建一个 Subject<IBaseFrame>会提供我需要的东西，我会让我的单线程从硬件接口(interface)读取数据，然后调用我的 Subject<IBaseFrame> 的 OnNext 函数实例。然后，不同的 IObserver 组件将从该主题接收通知。

我的困惑来自 this tutorial 附录中给出的建议它说:

Avoid the use of the subject types. Rx is effectively a functional programming paradigm. Using subjects means we are now managing state, which is potentially mutating. Dealing with both mutating state and asynchronous programming at the same time is very hard to get right. Furthermore, many of the operators (extension methods) have been carefully written to ensure that correct and consistent lifetime of subscriptions and sequences is maintained; when you introduce subjects, you can break this. Future releases may also see significant performance degradation if you explicitly use subjects.

我的应用程序对性能非常关键，我显然要在进入生产代码之前测试使用 Rx 模式的性能；然而，我担心我正在使用 Subject 类做一些违背 Rx 框架精神的事情，并且该框架的 future 版本会损害性能。

有没有更好的方法来做我想做的事？无论是否有任何观察者，硬件轮询线程都将持续运行(否则硬件缓冲区将备份)，因此这是一个非常热的序列。然后我需要将接收到的帧传递给多个观察者。

如有任何建议，我们将不胜感激。

Answer 1

好的，如果我们忽略我的教条方式并一起忽略“主题好/坏”。让我们看看问题空间。

我敢打赌，您需要融入 2 种风格的系统中的一种。

1. 系统在消息到达时引发事件或回调
2. 您需要轮询系统以查看是否有任何消息需要处理

对于选项 1，很简单，我们只需使用适当的 FromEvent 方法包装它就可以了。去酒吧!

对于选项 2，我们现在需要考虑如何轮询以及如何有效地执行此操作。还有当我们拿到值的时候，我们怎么发布呢？

我想你会想要一个专门的线程来进行轮询。您不希望其他编码员敲打 ThreadPool/TaskPool 并让您处于 ThreadPool 饥饿状态。或者你不想要上下文切换的麻烦(我猜)。所以假设我们有自己的线程，我们可能会有某种 While/Sleep 循环，我们坐在其中进行轮询。当检查发现一些消息时，我们会发布它们。好吧，所有这些听起来都非常适合 Observable.Create。现在我们可能不能使用 While 循环，因为它不允许我们返回 Disposable 以允许取消。幸运的是，您已经阅读了整本书，因此精通递归调度!

我想这样的事情是可行的。 #NotTested

public class MessageListener
{
    private readonly IObservable<IMessage> _messages;
    private readonly IScheduler _scheduler;

    public MessageListener()
    {
        _scheduler = new EventLoopScheduler();

        var messages = ListenToMessages()
                                    .SubscribeOn(_scheduler)
                                    .Publish();

        _messages = messages;
        messages.Connect();
    }

    public IObservable<IMessage> Messages
    {
        get {return _messages;}
    }

    private IObservable<IMessage> ListenToMessages()
    {
        return Observable.Create<IMessage>(o=>
        {
                return _scheduler.Schedule(recurse=>
                {
                    try
                    {           
                        var messages = GetMessages();
                        foreach (var msg in messages)
                        {
                            o.OnNext(msg);
                        }   
                        recurse();
                    }
                    catch (Exception ex)
                    {
                        o.OnError(ex);
                    }                   
                });
        });
    }

    private IEnumerable<IMessage> GetMessages()
    {
         //Do some work here that gets messages from a queue, 
         // file system, database or other system that cant push 
         // new data at us.
         // 
         //This may return an empty result when no new data is found.
    }
}

我真的不喜欢 Subjects 的原因是，这通常是开发人员没有真正对问题进行清晰设计的情况。破解一个主题，到处乱戳，然后让可怜的支持开发者猜测 WTF 正在进行中。当您使用 Create/Generate 等方法时，您是在将效果本地化到序列上。您可以在一种方法中看到这一切，并且您知道没有其他人会带来令人讨厌的副作用。如果我看到一个主题字段，我现在必须去寻找它正在使用的类中的所有位置。如果某个 MFer 公开暴露了一个，那么所有的赌注都会被取消，谁知道这个序列是如何被使用的!异步/并发/Rx 很难。您不需要通过允许副作用和因果关系编程让您的头脑更加困惑来让它变得更难。