c# - 在我的 C# 应用程序中结合使用 Task 和 IObservable 是一种不好的做法吗？

Question

我最近接触了 Rx，我正在使用它来帮助我从数据挖掘应用程序中的多个 API 中提取数据。

我有一个为每个 API 实现的接口(interface)，它封装了对每个 API 的常见调用，例如

public interface IMyApi {

    IObservable<string> GetApiName(); //Cold feed for getting the API's name.

    IObservable<int> GetNumberFeed(); //Hot feed of numbers from the API

}

我的问题是关于冷 IObservables 与任务。在我看来，冷 Observable 基本上是一项任务，它们的运行方式大致相同。当您可以争辩说您只需要一个任务时，将一个任务“抽象”为一个冷的可观察对象让我感到很奇怪。此外，使用冷可观察对象来包装任务隐藏了事件的性质，因为签名看起来与热可观察对象相同。

我可以表示上述接口(interface)的另一种方式是:

public interface IMyApi {

    Task<string> GetApiNameAsync(); //Async method for getting the API's name.

    IObservable<int> GetNumberFeed(); //Hot feed of numbers from the API

}

关于为什么我不应该在 Tasks 和 IObservables 之间混合搭配，是否有一些传统观点？

编辑:澄清一下——我已经阅读了发布的其他讨论并了解 Rx 和 TPL 之间的关系，但我主要关心的是在应用程序中将两者结合起来是否安全以及它是否会导致不良做法或线程和调度陷阱？

Answer 1

My question is around cold IObservables vs Tasks. In my mind, a cold observable is basically a task, they operate in much the same way

需要注意的是，不是这种情况，它们是非常不同的。这是核心区别:

// Nothing happens here at all! Just like calling Enumerable.Range(0, 100000000)
// doesn't actually create a huge array, until I use foreach.
var myColdObservable = MakeANetworkRequestObservable();

// Two network requests made!
myColdObservable.Subscribe(x => /*...*/);
myColdObservable.Subscribe(x => /*...*/);

// Only ***one*** network request made, subscribers share the
// result
var myTaskObservable = MakeATask().ToObservable();
myTaskObservable.Subscribe(x => /*...*/);
myTaskObservable.Subscribe(x => /*...*/);

为什么这很重要？ Rx 中的一些方法(例如 Retry)依赖于此行为:

// Retries three times, then gives up
myColdObservable.Retry(3).Subscribe(x => /*...*/);

// Actually *never* retries, and is effectively the same as if the
// Retry were never there, since all three tries will get the same
// result!
myTaskObservable.Retry(3).Subscribe(x => /*...*/);

所以一般来说，让你的 Observables 变得冷通常会让你的生活更轻松。

如何让任务冷？

使用 Defer 运算符:

var obs = Observable.Defer(() => CreateATask().ToObservable());

// CreateATask called *twice* here
obs.Subscribe(/*...*/);
obs.Subscribe(/*...*/);