TPL 与反应式框架

Question

什么时候会选择使用 Rx 而不是 TPL 或者这两个框架是正交的？

据我了解，Rx 的主要目的是提供对事件的抽象并允许组合，但它也允许提供对异步操作的抽象。 通过处理返回的 IDisposable 使用 Createxx 重载和 Fromxxx 重载和取消。

TPL 还通过任务和取消功能为操作提供抽象。

我的难题是何时使用哪个以及用于什么场景？

Answer 1

Rx 的主要目的不是提供对事件的抽象。这只是其结果之一。其主要目的是为集合提供可组合的推送模型。

反应式框架 (Rx) 基于 IObservable<T> 是 IEnumerable<T> 的数学对偶。因此，我们可以通过IObservable<T> 将对象“推送”给我们，而不是使用IEnumerable<T> 从集合中“拉取”项目。

当然，当我们真正去寻找可观察的来源时，诸如事件和异步操作之类的东西是极好的候选者。

响应式框架自然需要一个多线程模型，以便能够观察可观察数据的来源并管理查询和订阅。 Rx 实际上大量使用了 TPL 来执行此操作。

因此，如果您使用 Rx，那么您就是在隐式使用 TPL。

如果您希望直接控制您的任务，您可以直接使用 TPL。

但是，如果您有希望观察和执行查询的数据源，那么我完全推荐反应式框架。

Answer 2

我喜欢遵循的一些准则：

• 我处理的数据不是我自己产生的吗？数据随心所欲地到达？然后接收。
• 我是发起计算并且需要管理并发吗？然后是 TPL。
• 我是否管理多个结果，并且需要根据时间从它们中进行选择？然后接收。

Answer 3

2016 年 12 月更新：如果您有 30 分钟的时间，我建议您阅读 Joe Duffy 的第一手资料，而不是我的猜测。我认为我的分析很好，但如果你发现了这个问题，我强烈建议你看博文而不是这些答案，因为除了 TPL 与 Rx.NET 之外，他还涵盖了 MS 研究项目（Midori、Cosmos）。

http://joeduffyblog.com/2016/11/30/15-years-of-concurrency/

---

我认为 MS 在 .NET 2.0 出现后过度纠正是一个很大的错误。他们同时从公司的不同部门引入了许多不同的并发管理 API。

• Steven Toub 一直在努力推动线程安全原语替换 Event（从 Future<T> 开始，后来变成 Task<T>）
• MS Research 拥有 MIN-LINQ 和反应式扩展 (Rx)
• 硬件/嵌入式有robotics cuntime (CCR)

与此同时，许多托管 API 团队都在尝试使用 APM 和 Threadpool.QueueUserWorkItem()，但不知道 Toub 是否会赢得他在 mscorlib.dll 中发布 Future<T>/Task<T> 的斗争。最后，看起来他们对冲了，并在 mscorlib 中同时发布了 Task<T> 和 IObservable<T>，但不允许在 mscorlib 中使用任何其他 Rx API（甚至 ISubject<T>）。我认为这种对冲最终会导致大量重复（稍后会更多）并浪费公司内外的努力。

有关重复，请参阅：Task 与 IObservable<Unit>、Task<T> 与 AsyncSubject<T>、Task.Run() 与 Observable.Start()。而这只是冰山一角。但在更高的层次上考虑：

• StreamInsight - SQL 事件流，本机代码优化，但事件查询使用 LINQ 语法定义
• TPL 数据流 - 基于 TPL 构建，与 Rx 并行构建，针对调整线程并行性进行了优化，不擅长编写查询
• Rx - 惊人的表现力，但充满危险。将“热”流与IEnumerable 样式的扩展方法混合，这意味着您很容易永远阻塞（在热流上调用First() 永远不会返回）。调度限制（限制并行性）是通过相当奇怪的SubscribeOn() 扩展方法完成的，这些方法非常隐含且难以正确处理。如果开始学习 Rx，请预留很长的时间来学习所有要避免的陷阱。但如果组合复杂的事件流或您需要复杂的过滤/查询，Rx 确实是唯一的选择。

在 MS 在 mscorlib 中发布 ISubject<T> 之前，我认为 Rx 没有机会被广泛采用。这很可悲，因为 Rx 包含一些非常有用的具体（通用）类型，例如 TimeInterval<T> 和 Timestamped<T>，我认为它们应该在 Core/mscorlib 中，例如 Nullable<T>。另外，System.Reactive.EventPattern<TEventArgs>。

Answer 4

我喜欢 Scott W 的要点。放一些更具体的例子 Rx 非常好地映射到

• 消费流
• 执行非阻塞异步工作，例如网络请求。
• 流式事件（.net 事件，如鼠标移动或服务总线消息类型事件）
• 将事件的“流”组合在一起
• Linq 样式操作
• 从您的公共 API 公开数据流

TPL 似乎很好地映射到

• 工作的内部并行化
• 执行非阻塞异步工作，例如网络请求
• 执行工作流程和延续

我注意到 IObservable (Rx) 的一件事是它变得无处不在。一旦进入你的代码库，因为它无疑会通过其他接口公开，它最终会出现在你的应用程序中。我想这起初可能会很可怕，但大多数团队现在都对 Rx 感到非常满意，并且喜欢它为我们节省的工作量。

恕我直言，Rx 将成为 TPL 的主要库，因为 .NET 3.5、4.0、Silverlight 3、Silverlight 4 和 Javascript 已经支持它。这意味着您实际上必须学习一种风格，并且它适用于许多平台。

编辑：我改变了关于 Rx 优于 TPL 的想法。它们解决了不同的问题，因此不应该像那样进行比较。在 .NET 4.5/C# 5.0 中，async/await 关键字将进一步将我们与 TPL 联系起来（这很好）。有关 Rx 与事件与 TPL 等的深入讨论，请查看我的在线书籍 IntroToRx.com 中的 first chapter

Answer 5

我会说 TPL 数据流涵盖了 Rx 中专门的功能子集。 Dataflow 用于可能需要大量时间的数据处理，而 Rx 用于处理时间可以忽略不计的事件，例如鼠标位置、错误状态等。

示例：您的“订阅”处理程序是异步的，并且您当时希望不超过 1 个执行程序。使用 Rx 你必须阻止它，没有其他方法可以解决它，因为 Rx 与异步无关，并且在许多地方不会以特殊方式威胁异步。

.Subscribe(myAsyncHandler().Result)

如果你不阻塞，那么 Rx 会认为动作已经完成，而处理程序仍在异步执行。

如果你这样做，你可能会认为

.ObserveOn(Scheduler.EventLoopSchedule)

问题解决了。但这会破坏您的 .Complete() 工作流程，因为 Rx 会认为它在安排执行后立即完成，您将退出应用程序而无需等待异步操作完成。

如果您希望允许不超过 4 个并发异步任务，那么 Rx 不会提供任何开箱即用的功能。也许您可以通过实现自己的调度程序、缓冲区等来破解某些东西。

TPL Dataflow 在 ActionBlock 中提供了非常好的解决方案。它可以将同时操作限制到一定数量，并且它确实理解异步操作，因此调用 Complete() 并等待 Completed 将完全符合您的预期：等待所有正在进行的异步任务完成。

TPL 的另一个功能是“背压”。假设您在处理程序中发现了一个错误，需要重新计算上个月的数据。如果您使用 Rx 订阅源，并且您的管道包含无限缓冲区或 ObserveOn，那么您将在几秒钟内耗尽内存，因为源将继续读取速度超过处理可以处理的速度。即使您实现了阻塞消费者，您的源也可能会受到阻塞调用的影响，例如，如果源是异步的。在 TPL 中，您可以将源代码实现为

while(...)
    await actionBlock.SendAsync(msg)

在处理程序重载时不会阻塞源代码。

总的来说，我发现 Rx 非常适合时间和计算量小的动作。如果处理时间变得很长，那么您将处于奇怪的副作用和深奥的调试世界中。

好消息是 TPL 数据流块与 Rx 配合得非常好。它们具有 AsObserver/AsObservable 适配器，您可以在需要时将它们粘贴在 Rx 管道的中间。但是 Rx 有更多的模式和用例。所以我的经验法则是从 Rx 开始，根据需要添加 TPL 数据流。