立即交付第一个项目，“去抖动”以下项目答案

【问题标题】：Deliver the first item immediately, 'debounce' following items立即交付第一个项目，“去抖动”以下项目
【发布时间】：2015-07-20 07:50:32
【问题描述】：

考虑以下用例：

需要尽快交付第一件商品
需要在 1 秒超时后去抖动以下事件

我最终实现了基于OperatorDebounceWithTime 的自定义运算符，然后像这样使用它

.lift(new CustomOperatorDebounceWithTime<>(1, TimeUnit.SECONDS, Schedulers.computation()))

CustomOperatorDebounceWithTime 立即交付第一个项目，然后使用OperatorDebounceWithTime 运算符的逻辑去抖动 后面的项目。

是否有更简单的方法来实现所描述的行为？让我们跳过compose 运算符，它不能解决问题。我正在寻找一种在不实现自定义运算符的情况下实现此目的的方法。

【问题讨论】：

标签： rx-java rx-android

【解决方案1】：

更新：
来自@lopar 的 cmets 更好的方法是：

Observable.from(items).publish(publishedItems -> publishedItems.limit(1).concatWith(publishedItems.skip(1).debounce(1, TimeUnit.SECONDS)))

这样的工作是否可行：

String[] items = {"one", "two", "three", "four", "five", "six", "seven", "eight"};
Observable<String> myObservable = Observable.from(items);
Observable.concat(myObservable.first(), myObservable.skip(1).debounce(1, TimeUnit.SECONDS))
    .subscribe(s -> System.out.println(s));

【讨论】：

您还可以使用publish：Observable.from(items).publish(publishedItems -> publishedItems.limit(1).concatWith(publishedItems.skip(1).debounce(1, TimeUnit.SECONDS))) 防止进行双重订阅（在可观察的冷情况下，它会执行两次工作，而在热情况下可能会不同步）如果Observable 为空，first 将爆炸，因此除非这是所需的行为，否则您可能需要limit(1)。
我在等待@lopar 提出的解决方案。我认为第二个可能在某些情况下不起作用。 myObservable.skip(1) 可以在第一个项目发布后订阅，因此它会跳过第二个元素。
@LordRaydenMK，你原来的答案很好，初学者不用担心！ :) 很高兴我能帮忙调整一下。
使用发布表单时，您不应该.skip(1)。因为 observable 使用的是同一个订阅，所以它会从同一个位置继续。它应该只是Observable.from(items).publish(publishedItems -> publishedItems.limit(1).concatWith(publishedItems.debounce(1, TimeUnit.SECONDS)))
在版本 2 中使用 .take(1) 而不是 .limit(1)

【解决方案2】：

@LortRaydenMK 和 @lopar 的答案是最好的，但我想提出其他建议，以防它恰好对您或处于类似情况的人更有效。

debounce() 有一个变体，它采用一个函数来决定该特定项目去抖动多长时间。它通过返回一个在一段时间后完成的 observable 来指定这一点。您的函数可以为第一项返回empty()，为其余的返回timer()。类似（未经测试）：

String[] items = {"one", "two", "three", "four", "five", "six"};
Observable.from(items)
    .debounce(item -> item.equals("one")
            ? Observable.empty()
            : Observable.timer(1, TimeUnit.SECONDS));

诀窍是这个函数必须知道哪个项目是第一个。你的序列可能知道这一点。如果没有，您可能必须 zip() 和 range() 或其他东西。在这种情况下，最好使用其他答案中的解决方案。

【讨论】：

应该有 TimeUnit.SECONDS（你错过了“S”）。但是，这是我发现的唯一使用此功能的示例。它真的帮助了我。非常感谢！

【解决方案3】：

使用 RxJava 2.0 的简单解决方案，翻译自 the answer for the same question for RxJS，结合了throttleFirst 和 debounce，然后删除重复项。

private <T> ObservableTransformer<T, T> debounceImmediate() {
    return observable  -> observable.publish(p -> 
        Observable.merge(p.throttleFirst(1, TimeUnit.SECONDS), 
            p.debounce(1, TimeUnit.SECONDS)).distinctUntilChanged());
} 

@Test
public void testDebounceImmediate() {
    Observable.just(0, 100, 200, 1500, 1600, 1800, 2000, 10000)
        .flatMap(v -> Observable.timer(v, TimeUnit.MILLISECONDS).map(w -> v))
        .doOnNext(v -> System.out.println(LocalDateTime.now() + " T=" + v))
            .compose(debounceImmediate())
            .blockingSubscribe(v -> System.out.println(LocalDateTime.now() + " Debounced: " + v));
}

使用 limit() 或 take() 的方法似乎无法处理长期存在的数据流，我可能希望持续观察，但仍会在一段时间内看到第一个事件时立即采取行动。

【讨论】：

【解决方案4】：

LordRaydenMK and lopar's answer 有一个问题：你总是丢失第二个项目。我想以前没有人意识到这一点，因为如果你有一个去抖动，你通常会有很多事件，而第二个事件无论如何都会被去抖动。永不丢失事件的正确方法是：

observable
    .publish(published ->
        published
            .limit(1)
            .concatWith(published.debounce(1, TimeUnit.SECONDS)));

别担心，你不会得到任何重复的事件。如果您不确定，您可以运行此代码并自行检查：

Observable.just(1, 2, 3, 4)
    .publish(published ->
        published
            .limit(1)
            .concatWith(published))
    .subscribe(System.out::println);

【讨论】：

【解决方案5】：

使用带函数的debounce的版本，这样实现函数：

    .debounce(new Func1<String, Observable<String>>() {
        private AtomicBoolean isFirstEmission = new AtomicBoolean(true);
        @Override
        public Observable<String> call(String s) {
             // note: standard debounce causes the first item to be
             // delayed by 1 second unnecessarily, this is a workaround
             if (isFirstEmission.getAndSet(false)) {
                 return Observable.just(s);
             } else {
                 return Observable.just(s).delay(1, TimeUnit.SECONDS);
             }
        }
    })

第一个项目立即发出。后续项目延迟一秒。如果延迟的 observable 没有在下一个项目到达之前终止，它就会被取消，所以预期的去抖动行为就实现了。

【讨论】：

【解决方案6】：

基于@lopar 评论的 Kotlin 扩展函数：

fun <T> Flowable<T>.debounceImmediate(timeout: Long, unit: TimeUnit): Flowable<T> {
    return publish {
        it.take(1).concatWith(it.debounce(timeout, unit))
    }
}

fun <T> Observable<T>.debounceImmediate(timeout: Long, unit: TimeUnit): Observable<T> {
    return publish {
        it.take(1).concatWith(it.debounce(timeout, unit))
    }
}

【讨论】：

【解决方案7】：

Ngrx - rxjs 解决方案，将管道一分为二

onMyAction$ = this.actions$
    .pipe(ofType<any>(ActionTypes.MY_ACTION);

lastTime = new Date();

@Effect()
onMyActionWithAbort$ = this.onMyAction$
    .pipe(
        filter((data) => { 
          const result = new Date() - this.lastTime > 200; 
          this.lastTime = new Date(); 
          return result; 
        }),
        switchMap(this.DoTheJob.bind(this))
    );

@Effect()
onMyActionWithDebounce$ = this.onMyAction$
    .pipe(
        debounceTime(200),
        filter(this.preventDuplicateFilter.bind(this)),
        switchMap(this.DoTheJob.bind(this))
    );

【讨论】：

【解决方案8】：

为了防止双重订阅使用这个：

    const debouncedSkipFirstStream$ = stream$.pipe(
        map((it, index) => ({ it, index })),
        debounce(({ index }) => (
            index ? new Promise(res => setTimeout(res, TimeUnit.SECONDS))
                : Rx.of(true))),
        map(({ it }) => it),
    );

如果使用拆分解决方案，您将看到“运行”打印两次

x = rxjs.Observable.create(o=>{
    console.info('run');
    o.next(1);
    o.next(2);
});
a = x.pipe(rxjs.operators.take(1));
b = x.pipe(rxjs.operators.skip(1), rxjs.operators.debounceTime(60));
rxjs.concat(a, b).subscribe(console.log);

【讨论】：

【解决方案9】：

我对 Dart 的解决方案：

extension StreamExt<T> on Stream<T> {
  Stream<T> immediateDebounce(Duration duration) {
    var lastEmit = 0;
    return debounce((event) {
      if (_now - lastEmit < duration.inMilliseconds) {
        lastEmit = _now;
        return Stream.value(event).delay(duration);
      } else {
        lastEmit = _now;
        return Stream.value(event);
      }
    });
  }
}

int get _now =>  DateTime.now().millisecondsSinceEpoch;

【讨论】：

【解决方案10】：

我去了

Flowable.concat(

    flowable // emits immediately
        .take(1)
        .skipWhile { it.isEmpty() },

    flowable // same flowable, but emits with delay and debounce
        .debounce(2, TimeUnit.SECONDS)
)
    .distinctUntilChanged()

【讨论】：

【解决方案11】：

如果有人在 2021 年寻找这个：

@OptIn(FlowPreview::class)
fun <T> Flow<T>.debounceImmediate(timeMillis: Long): Flow<T> =
    withIndex()
        .onEach { if (it.index != 0) delay(timeMillis) }
        .map { it.value }

用法：

authRepository.login(loginDto)
                    .debounceImmediate(10000)

【讨论】：

【解决方案12】：

阅读this article 后，我最终使用throttleLatest 运算符获得与我正在寻找的立即去抖动非常相似的行为。

以下代码将立即发出第一个项目，然后每 500 毫秒检查一次新项目。只会发送在该 500 毫秒窗口内收到的最新事件。

observable.throttleLatest(500, TimeUnit.MILLISECONDS)

【讨论】：

【解决方案13】：

   view.clicks()
            .throttleFirst(2, TimeUnit.SECONDS)
            .subscribe {
                println("Clicked button")
            }

我发现这是最简单的方法。 clicks() 来自 rx 视图绑定。添加此依赖以获得可观察的视图

 implementation 'com.jakewharton.rxbinding4:rxbinding:4.0.0

【讨论】：

虽然此代码可能会回答问题，但提供有关它如何和/或为什么解决问题的额外上下文将提高答案的长期价值。

【解决方案14】：

对于那些试图使用 Kotlin Flow 解决相同问题的人：

fun <T> Flow<T>.throttleFirst(timeout: Duration): Flow<T> {
    var job = Job().apply { complete() }
    return onCompletion { job.cancel() }.run {
        flow {
            coroutineScope {
                collect { value ->
                    if (!job.isActive) {
                        emit(value)
                        job = launch { delay(timeout.inWholeMilliseconds) }
                    }
                }
            }
        }
    }
}

例子：

flow {
    emit(1)
    delay(90)
    emit(2)
    delay(90)
    emit(3)
    delay(1010)
    emit(4)
    delay(1010)
    emit(5)
}.throttleFirst(1.seconds).collect { ... }
// 1, 4, 5

Ref

【讨论】：