rx-java - 如何过滤变化超过增量阈值的值

Question

我有一个发出数字的 Observable(整数、 double 、BigDecimal 实际上并不重要)。我在可观察的合成时间定义了一个阈值。换句话说，阈值在可观察的生命周期内不会改变。

我需要的是过滤掉与阈值内最后通过的值不同的值。或者改写一下:只有当值与最后传递的值的差异大于阈值时，才向下游传递值。

示例会更好地解释它:

Source observable values: 
[5, 4, 7, 9, 15, 14, 13, 12, 11, 7, 3, 2, 1]

Filetered values with threshold = 3 : 
[5, 9, 15, 11, 7, 3]

为了更好地说明这一点，我们假设它是温度读数。我只想过滤那些温度变化超过 3 度的值。第一个值始终会传递并用作要比较的初始值。例如，如果第一个温度读数为 21 摄氏度，并且在一小时内每个后续读数都在 [18..24] 中，那么这些值都不应传递到下游。但一旦超过这些界限，就应该通过并创建新的界限进行比较。

问题是:如何仅使用 RxJava 运算符来做到这一点？或者除了在 r​​x 管道之外存储状态之外没有其他方法(在 volatile 或原子引用或任何其他同步状态中，细节并不重要)？

Answer 1

您可以结合使用 scan 和 distinctUntilChanged 只需 RX 运算符即可实现此目的。

scan 允许您访问上一个和下一个值，以便您与给定的阈值进行比较。

如果未达到阈值，那么您可以只发出之前的值。

然后可以使用

distinctUntilChanged来删除重复的排放。

在 Kotlin 中，这将如下所示:

val threshold = 3
val list = listOf(5, 4, 7, 9, 15, 14, 13, 12, 11, 7, 3, 2, 1)

return Observable.fromIterable<Int>(list)
        .scan { previous: Int, next: Int ->
            val difference = abs(next - previous)
            when {
                difference > threshold -> next
                else -> previous
            }
        }.distinctUntilChanged()