如何延迟项目，但只在开始时延迟一次？

Question

delay运算符将所有项延迟指定的时间量。我只想延迟和缓冲前N秒的项目。在N秒之后，应该不会有延迟。我需要在下面的代码中做到这一点。

private Emitter<Work> workEmitter;

// In the constructor.
Flowable.create(
        (FlowableOnSubscribe<Work>) emitter -> workEmitter = emitter.serialize(),
        BackpressureStrategy.BUFFER)
    .observeOn(Schedulers.from(executor))
    .subscribe(work -> process(work));

// On another thread, as work comes in, ...
workEmitter.onNext(t);

我想要做的是在前N秒内推迟工作的处理，但不是在那之后。我尝试了delaySubscription，但它在延迟期间将workEmitter保留为null。我之所以要这样做，是为了让CPU在最初阶段可用于其他重要工作。

Answer 1

您可以使用UnicastProcessor并在一段延迟后订阅它：

FlowableProcessor<Work> processor = UnicastProcessor.<Work>create().toSerialized();

processor.delaySubscription(N, TimeUnit.SECONDS)
.observeOn(Schedulers.from(executor))
.subscribe( work -> process(work));

// On another thread, as work comes in, ...
processor.onNext(t);

UnicastProcessor将保持对工作项的缓冲，直到delaySubscription的时间过去，然后切换到它。

Answer 2

您可以延迟创建可观察对象，然后订阅它。

Observable.timer( N, SECONDS )
  .flatMap( ignored -> Flowable.create(
    (FlowableOnSubscribe<Work>) emitter -> workEmitter = emitter.serialize(),
       BackpressureStrategy.BUFFER)
    .observeOn(Schedulers.from(executor)))
  .subscribe( work -> process(work));

直到经过N秒之后，观察者链才会启动。