Observable.Range 是否会破坏 Observable 合约？

Question

在了解 Rx 的过程中，我遇到了一个经常重复的关于 Observables 的规则，在 The Observable Contract 中有详细说明。

在发出 OnCompleted 或 OnError 通知后，它可能不会再发出任何进一步的通知。

这对我来说很有意义，因为让 Observable 在完成后继续产生值会令人困惑，但是当我在 .NET 中测试 Observable.Range 方法时，我注意到它没有表现出这种行为，事实上很多 Observables 违反了这条规则。

var rangeObservable = Observable.Range(0, 5);

rangeObservable.Subscribe(Console.WriteLine, () => Console.WriteLine("Done first!"));
Console.ReadLine();

rangeObservable.Subscribe(Console.WriteLine, () => Console.WriteLine("Done second!"));
Console.ReadLine();

//Output:
//0
//1
//2
//3
//4
//Done first!

//0
//1
//2
//3
//4
//Done second!

显然rangeObservable 调用了两次OnComplete 并在第一次OnComplete 之后产生了值。这让我相信这不是关于 Observables 的规则，而是关于 Subscriptions 的规则。也就是说，一个 Observable 可以产生任意多的终止消息，甚至在它产生之后产生值，只要每个 Subscription 只接收一个终止消息并且没有收到之后的进一步消息。

当它说 Observable 时，它们实际上是指 Subscription 吗？它们真的是不同的东西吗？我对模型有根本的误解吗？

Answer 1

observable 合约必须对任何被观察的 Observable 有效。 在 Observable 未被观察时是否发生任何事情都留给 observable 的实现。

在 Enumerable 中考虑类比会有所帮助 - Observable 是 Enumerable 的对偶。在可枚举中，你会有 range = Enumerable.Range(0, 5)，您将使用与上述类似的范围：

range.ForEach(Console.WriteLine); //prints 0 - 4

range.ForEach(Console.WriteLine); //prints 0 - 4 again

并发现这是完全可以接受的行为，因为只有在调用 GetEnumerator 时才会创建实际的数字生成器。同样，在 Observable 中，等价的方法是Subscribe。

范围的实现是这样的：

        static IObservable<int> Range(int start, int count)
        {
            return Observable.Create<int>(observer =>
            {
                for (int i = 0; i < count; i++)
                    observer.OnNext(start + i);

                observer.OnCompleted();

                return Disposable.Empty;
            });
        }

这里，每次有订阅时都会调用observer => {...} 函数。工作在 subscribe 方法中完成。您可以很容易地看到它 (1) 为每个观察者推送相同的序列，(2) 每个观察者只完成一次。

这些只有在你观察它们时才会发生某些事情的可观察对象称为冷可观察对象。 Here's an article 描述这个概念。

注意

Range 是一个非常幼稚的实现，仅用于说明目的。该方法在完成之前不会返回一次性用品 - 所以Disposable.Empty 是可以接受的。正确的实现将在调度程序上运行工作，并使用已检查的一次性来查看订阅是否已被释放，然后再继续循环。

要点是手动实现可观察合约是困难，这就是存在 Rx 库的原因 - 通过组合构建功能。

Answer 2

Observable.Range 返回一个 cold 可观察对象，这意味着它为每个订阅者“重播”它的行为。由于“OnNext* OnComplete|OnError”合约仅适用于订阅，这完全没问题。

有关热/冷 observables 的更多信息，请参阅my answer on "IConnectableObservables in Rx"