java - RxJava Observable/Flowable 上的运算符延迟 n 个项目的发射

Question

我想转换Flowable，以便它推迟发射项目，直到收集到指定数量的项目，然后按照 FIFO 顺序将它们发射到下游，保持恒定的延迟项目计数。一旦上游发出 onComplete 信号，排队的项目应该在发出 onComplete 之前刷新到下游:

(在此示例中，延迟的项目编号为 3)

1 2 3 4 5 6 7 |
      1 2 3 4 5 6 7 |

我没有看到任何现有的运算符可以执行此操作或可以进行修改以获得该行为。 Observable.delay 似乎仅支持基于时间的延迟，而不支持基于计数的延迟。

实现自定义运算符来实现此目的应该很容易，但也许现有运算符有更简单的方法？

Answer 1

您可以发布一个序列，跳过最后 N 个，然后将最后 N 个追加回来:

Flowable.range(1, 7)
    .flatMap(v -> Flowable.timer(v * 200, TimeUnit.MILLISECONDS).map(w -> v))
    .doOnNext(v -> System.out.println(v))
// -------------------------------------------------------------------
    .publish(f -> 
        f.skipLast(3).mergeWith(f.takeLast(3))
    )
// -------------------------------------------------------------------
    .blockingSubscribe(v -> System.out.println("<-- " + v));