c# - 使用 Subject 解耦 Observable 的订阅和初始化

Question

我有一个公开IObservable 状态的API。但此状态取决于必须通过 Init 初始化的底层可观察源。

我想做的是保护用户不必按正确的顺序做事:就目前而言，如果他们在执行 之前尝试订阅 Status Init，他们得到一个异常，因为他们的源没有被初始化。

所以我有了一个天才的想法，使用 Subject 将两者分离:订阅我的 Status 的外部用户只是订阅了 Subject，然后当他们调用Init，我使用我的主题订阅基础服务。

代码中的想法

private ISubject<bool> _StatusSubject = new Subject<bool>();
public IObservable<bool> Status { get { return _StatusSubject; } }

public void Init() 
{
    _Connection = new Connection();
    Underlying.GetDeferredObservable(_Connection).Subscribe(_StatusSubject);
}

但是，从虚拟项目的测试来看，问题在于初始化通过订阅 Subject 来“唤醒”我的底层 Observable，即使还没有人订阅该主题。如果可能的话，我想避免这种情况，但我不确定如何...

(我也注意到 the received wisdom，“一般规则是，如果您使用的是主题，那么您就做错了”)

Answer 1

您似乎缺少的概念是如何知道有人何时开始收听并仅初始化您的基础源。通常您使用 Observable.Create 或其兄弟之一(Defer、Using、...)来执行此操作。

下面是没有 Subject 的方法:

private IObservable<bool> _status = Observable.Defer(() =>
{
    _Connection = new Connection();
    return Underlying.GetDeferredObservable(_Connection);
};

public IObservable<bool> Status { get { return _status; } }

Defer 在有人实际订阅之前不会调用初始化代码。

但这有几个潜在的问题:

1. 每个观察者都会建立一个新的连接
2. 当观察者取消订阅时，连接不会被清除。

第二个问题很容易解决，所以让我们先解决这个问题。假设您的 Connection 是一次性的，在这种情况下您可以这样做:

private IObservable<bool> _status = Observable
    .Using(() => new Connection(),
           connection => Underlying.GetDeferredObservable(connection));

public IObservable<bool> Status { get { return _status; } }

通过此迭代，每当有人订阅时，都会创建一个新的 Connection 并将其传递给第二个 lamba 方法以构造可观察对象。每当观察者取消订阅时，Connection 就会被Disposed。如果 Connection 不是 IDisposable，那么您可以使用 Disposable.Create(Action) 创建一个 IDisposable，它将运行您需要运行的任何操作来清理连接。

您仍然遇到每个观察者都创建一个新连接的问题。我们可以使用 Publish 和 RefCount 来解决这个问题:

private IObservable<bool> _status = Observable
    .Using(() => new Connection(),
           connection => Underlying.GetDeferredObservable(connection))
    .Publish()
    .RefCount();

public IObservable<bool> Status { get { return _status; } }

现在，当第一个 观察者订阅时，将创建连接并订阅底层的可观察对象。随后的观察者将共享连接并获取当前状态。当最后一个 观察者取消订阅时，连接将被处理掉，一切都将关闭。如果之后有另一个观察者订阅，一切都会重新开始。

在幕后，Publish 实际上是使用 Subject 来共享单个可观察源。 RefCount 正在跟踪当前正在观察的观察者数量。