47

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

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

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

我的困境是什么时候使用哪个以及在什么场景下使用？

4

5 回答 5

50

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

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

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

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

因此，如果您使用 Rx，那么您就隐含地使用了 TPL。

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

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

于 2010-03-30T06:13:16.607 回答
27

我喜欢遵循的一些准则：

  • 我是否在处理不是我原创的数据。数据随心所欲地到达？然后接收。
  • 我是发起计算并需要管理并发吗？然后是TPL。
  • 我是否管理多个结果，并且需要根据时间从它们中进行选择？然后接收。
于 2010-04-21T23:45:40.007 回答
15

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 和 Reactive Extensions (Rx)
  • 硬件/嵌入式有机器人 cuntime (CCR)

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

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

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

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

于 2013-01-21T19:15:37.240 回答
13

我喜欢 Scott W 的要点。在 Rx 地图中放置一些更具体的示例非常好

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

TPL 似乎很好地映射到

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

我注意到 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的第一章

于 2010-07-16T08:17:12.083 回答
10

我会说 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 非常适合时间和计算量小的动作。如果处理时间变得很长，那么您将处于奇怪的副作用和深奥的调试世界中。

Good news is that TPL Dataflow blocks play very nice with Rx. They have AsObserver/AsObservable adapters and you can stick them in the middle of Rx pipeline when needed. But Rx has much more patterns and use cases. So my rule of thumb is to start with Rx and add TPL Dataflow as needed.

于 2015-04-13T20:12:57.373 回答