.net - 什么是 IObservable.Subscribe(IObserver observer) 重载？

Question

当我写 .Subscribe 时我经常发现 Resharper 为我选择了以下重载，位于 mscorlib 中，Version=4.0.0.0:

namespace System
{
  public interface IObservable<out T>
  {
    IDisposable Subscribe(IObserver<T> observer);
  }
}

这似乎与大多数采用 Action 的重载非常不同而且它来自 mscorlib 而不是 System.Reactive.*这是我期望大多数响应式内容的地方。

这个重载有什么作用？应该如何使用？IObserver<T>如何关于 Action ？为什么在 mscorlib 中会出现这种单一重载？

Answer 1

需要说明的是，这不是重载，而是 Rx 的真正核心。您习惯使用的所有其他 Subscribe 方法以及与此相关的所有其他运算符实际上都是最终调用它的扩展方法。

如果您查看早期文档或 Rx，您会发现创建者将其视为 LINQ 的基于推送的一面。所以很多东西都是你在 LINQ 中看到的东西的镜像。 IObservable是IEnumerable的镜像，IObserver是IEnumerator的镜像。

但是，因为推与拉相反，所以 Rx 版本与其基于拉的版本相反:

• IEnumerable 定义了一种生成 IEnumerator 的方法。 IObservable 定义了一种接收 IObserver 的方法。
• 如果您将 IEnumerator.MoveNext() + IEnumerator.Current 视为一个操作，则可以通过以下三种方式之一返回:返回下一个元素、集合结束达到，或抛出异常。同样，IObserver 必须处理三种情况:下一个元素 (OnNext)、流结束 (OnCompleted) 或异常 ( OnError).

Subscribe 的更熟悉的“重载”实际上只是看起来像这样的扩展方法:

public static IDisposable Subscribe<T>(this IObservable<T> source, Action<T> onNext)
{
    return Subscribe(onNext, e => {/*onError */}, () => {/*onCompleted*/);
}

public static IDisposable Subscribe<T>(this IObservable<T> source, Action<T> onNext, Action<Exception> onError, Action onCompleted)
{
    source.Subscribe(new AnonymousObserver<T>(onNext, onError, onCompleted));
}