这会是反应式扩展的良好候选者吗?

Question

我有一个需要提高性能的过程。

此过程在数据库中运行查询。

对于每条记录，根据记录中的数据，可能必须完成 2 个任务。

对于这两个任务之一，可能需要从 Web 服务获取其他信息，然后，处理完全相同。

另一个任务总是相同的。

完成这两项任务后，必须在数据库中更新记录。

我认为 Rx 可以让我并行运行进程，从而提高性能。另外，我希望能够监视批处理的进度。

目前，我有以下代码。我想知道这似乎是完成这项任务的有效方法。

IObservable<Record> records = GetDataFromDb().ToObservable();
var recsOfType1 = records.Where(r => r.IsOfType1).Do(FetchAdditionalInfoIntoRecord);
var recsOfType2 = records.Where(r => r.IsOfType2);
var allCompleteRecords = recsOfType1.Merge(recsOfType2);
allCompleteRecords.Do(Task1).Do(Task2).Do(UpdateInDB);

我需要添加额外的错误处理和对进程的监视。我不是100%确定我会如何处理。

欢迎任何想法/评论/批评。

谢谢

Answer 1

正如评论提到的，Rx 最适合基于推送的流。所有这些任务都是基于拉动的。如果你觉得标准任务不能削减它，那么我会推荐TPL-Dataflow。

Answer 2

数据库查询、文件、注册表等是基于拉取的（可枚举的），而不是基于推送的（可观察的）。您可以使用交互式扩展（System.Interactive）代替反应式扩展 - 它们包含 Rx 中的大多数等效方法，但基于拉取。

可枚举生成器允许您编写协程（通过yield），并且是一种强大的组合工具。如果需要异步，只需将枚举包装在任务中即可。