c# - 在没有 onBackpressureLatest 的情况下处理 Rx.NET 中的背压

Question

我需要在 Rx.NET 中实现以下算法:

1. 从 stream 中获取最新的项目，如果没有新项目，则不阻塞地等待新项目。只有最新的项目很重要，其他的可以放弃。
2. 将项目输入到 SlowFunction 并打印输出。
3. 从第 1 步开始重复。

天真的解决方案是:

let PrintLatestData (stream: IObservable<_>) =
    stream.Select(SlowFunction).Subscribe(printfn "%A")

但是，此解决方案不起作用，因为平均而言 stream 发出项目的速度比 SlowFunction 消耗它们的速度快。由于 Select 不会丢弃项目，而是尝试按从旧到新的顺序处理每个项目，因此在程序运行时，项目发出和打印之间的延迟将增长到无穷大。应仅从流中获取最新的最近项，以避免这种无限增长的背压。

我搜索了文档并在 RxJava 中找到了一个名为 onBackpressureLatest 的方法，根据我的理解，它可以执行我上面描述的操作。但是，该方法在 Rx.NET 中不存在。如何在 Rx.NET 中实现它？

Answer 1

我想你想使用像 ObserveLatestOn 这样的东西。它有效地用单个值和一个标志替换了传入事件队列。

James World 在这里发表了博客 http://www.zerobugbuild.com/?p=192

这个概念在 GUI 应用程序中大量使用，它们不能相信服务器向其推送数据的速度有多快。

您还可以在 Reactive Trader 中看到一个实现 https://github.com/AdaptiveConsulting/ReactiveTrader/blob/83a6b7f312b9ba9d70327f03d8d326934b379211/src/Adaptive.ReactiveTrader.Shared/Extensions/ObservableExtensions.cs#L64 以及解释 ReactiveTrader 的支持演示文稿 https://leecampbell.com/presentations/#ReactConfLondon2014

需要明确的是，这是一种负载卸载算法，而不是背压算法。