rxJava debounce() 运算符不适用于 Observable.range()

Question

AFAIK，rxJava 的debounce()运算符，用于延迟事件的发出。当我使用搜索框应用它时，它可以正常工作：

RxTextView.textChangeEvents(editText)
.debounce(1000, TimeUnit.MILLISECONDS) //Only emit after 1 sec
.subscribe(new Observer<TextViewTextChangeEvent>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(TextViewTextChangeEvent event) {
//Get more information about text change event
Log.e(TAG, "Before: " + event.before() + ", start: " + event.start() + ", count: " + event.count());
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
Log.e(TAG, "textChangeEvents: onComplete");
}
});

但是当我像这样Observable.range()应用它时：

Observable.range(1, 10000)
.debounce(1000, TimeUnit.MILLISECONDS)
.subscribe(new Observer<Long>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
}
@Override
public void onNext(@NonNull Long integer) {
}
@Override
public void onError(@NonNull Throwable e) {
}
@Override
public void onComplete() {
}
});

排放量onNext()非常快(大约 1000 次排放/秒)并且持续不断，尽管我已经申请了debounce(1000, TimeUnit.MILISECONDS)运算符。

我期待的是：当我使用debounce()延迟 1000 毫秒后只会发出 1000 个数字(在 2 次发射之间延迟时它可以跳过数字)。意味着排放将一个接一个地向下游移动，就像上面的搜索框示例一样。

我是 rx 的新手，请帮助我实现这一目标并解释原因？我真的不知道为什么我必须使用另一个运算符，但debounce()因为想法是一样的！

Answer 1

debounce通过定义一个宽限期来防止下游不堪重负，该宽限期必须在事件之间经过才能获得最后一个事件，换句话说，它将在一段时间安静后发出最新元素。范围将尽可能快地通过其项目，因此项目之间没有足够的时间，只会发出最后一个。

debounce根本不是您的用例所需的运算符。2.x 的扩展项目为您的用例提供了 spanout 运算符。

Answer 2

你需要的是每个项目'.delay()'运算符。请参阅此处的说明 - http://reactivex.io/RxJava/javadoc/rx/Observable.html#delay-rx.functions.Func1-

Observable
.range(1, 10000)
.delay(new Func1<Integer, Observable<Long>>() {
@Override
public Observable<Long> call(Integer integer) {
return Observable.timer(1, TimeUnit.SECONDS );
}
});

对于良好的顺序，".debounce()">对于这个用例来说绝对是毫无疑问的。

Answer 3

使用延迟而不是去抖动

Observable.range(1, 10000)
.delay(1000, TimeUnit.MILLISECONDS)
.subscribe(...);

Answer 4

Observable.range(1, 10000)
.debounce(500, TimeUnit.MILLISECONDS)
// Run on a background thread
.subscribeOn(Schedulers.io())
// Be notified on the main thread
.observeOn(AndroidSchedulers.mainThread())
.subscribe(....);

试试这种方式。