如何使用 .net RX 组织数据处理器序列

Question

使用 .net RX 组织数据处理器序列的最佳方法是什么？
- 一个。在 observable 上调用方法，例如 observable.Do(log).Select(transformation).Do(work).Aggregate(someState)...
- 乙。实现自定义观察者，如果是这样 - 如何链接它们
- C。其他选项.. 还有什么是处理 observable 本身中可能的异常的最佳选择（请参阅上面的问题）并处理 Do、Select 等内部的异常（我知道最佳实践是订阅者不应该抛出）。

我有时还需要允许异常作为可观察序列的某些元素返回而不停止序列（请参阅此问题Handling Exceptions in Reactive Extensions without stopping sequence）

Answer 1

这似乎您需要一个工作流程而不是 Rx。似乎基于您的其他问题*，您正在尝试采用看起来非常适合 ProducerConsumer 排队工作流场景的内容，并将其强制用于 Rx。

看起来像你

• 想要从队列中读取并阻塞直到接收到一个值，然后 重新订阅。只需使用 BlockingCollection 的排队功能。当值到达时，它们可以从任何线程推送到集合中。
• 不希望实际上有一个值序列，而是一个可以 在任何给定值上失败，转移它然后处理下一个值。 只需使用队列。处理每个值并将结果放入下一个适当的队列 （失败/成功）->见Enterprise Integration Patterns， 特别是无效的消息通道和死信通道。
• 可能会并行处理这些值。看 BlockingCollection.GetConsumingEnumerable 或 Disruptor 高性能实现。使用这些工具，您可以拥有许多生产者和消费者。当然，你可以用 Rx 做到这一点，但它只是轮询和绑定一个线程。我认为最好明确说明这种事情

我认为您对 Rx 的使用应该为自己付出代价，并且您不应该发现自己与它抗争（就像任何其他技术或框架一样）。

*其他问题，例如 Handling Exceptions in Reactive Extensions without stopping sequence 和 How to serialize Observables to the cloud and back