我正在尝试使用java反应式扩展构建一个简单的应用程序。我有两个连续发出温度值的流,我想检测并滤除可能出现错误的感测温度峰值,这样做我需要考虑先例值,以便我可以考虑到这样的变化:
我仍然无法在文档中找到合适的操作符。有谁知道我该如何完成任务?我应该自定义运营商吗?
这些是我的流:
double min = 50, max = 75, spikeFreq = 0.01;
Observable<Double> tempStream1 = Observable.create((
Subscriber<? super Double> subscriber) -> {
new TempStream(subscriber, min, max, spikeFreq).start();
});
Observable<Double> tempStream2 = Observable.create((
Subscriber<? super Double> subscriber) -> {
new TempStream(subscriber, min, max, spikeFreq).start();
});
public class TempStream extends Thread{
private Subscriber<? super Double> subscriber;
private TempSensor sensor;
public TempStream(Subscriber<? super Double> subscriber, double min,
double max, double spikeFreq) {
this.subscriber = subscriber;
sensor = new TempSensor(min, max, spikeFreq);
}
@Override
public void run() {
Random gen = new Random(System.currentTimeMillis());
while (!subscriber.isUnsubscribed()) {
try {
subscriber.onNext(sensor.getCurrentValue());
Thread.sleep(1000 + gen.nextInt() % 1000);
} catch (Exception ex) {
subscriber.onError(ex);
}
}
subscriber.onCompleted();
}
}
也许buffer
运算符(http://reactivex.io/documentation/operators/buffer.html)可能在这种情况下有所帮助。你想使用buffer
与count = 2
和skip = 1
。这样你就可以获得流中一个元素的“前瞻”。
例如:
stream.buffer(2,1).filter(buf -> buf.size() == 2 && buf.get(0) - buf.get(1) < max);
请注意,此示例还检查是否缓冲了两个值,因为在完成流时可能只发生一个值。