Flux.repeat() 不重复

Question

我正在尝试像这样在 Flux 上重复订阅：    DirectProcessor<String> stringDirectProcessor = DirectProcessor.create();     stringDirectProcessor             .repeat(3)             .subscribe(item -> System.out.println(item));     stringDirectProcessor.onNext("one");     stringDirectProcessor.onNext("two");     stringDirectProcessor.onNext("three");     stringDirectProcessor.onComplete();我的期望是看到这个输出：one two three one two three one two three one two three但我只得到one two three但是，如果我使用Flux.just()而不是DirectProcessor我得到预期的输出。怎么了？

Answer 1

这是 的预期行为DirectProcessor。我刚刚通读了文档，发现了以下有关的内容DirectProcessor：

一旦处理器终止（通常通过其接收器的 error(Throwable) 或 complete() 方法被调用），它允许更多订阅者订阅但立即向他们重播终止信号。

因此，由于repeat简单地重新订阅，onComplete处理程序将立即被调用。你确定你需要吗DirectProcessor？

编辑：这里也记录了这种行为

注意：如果没有订阅者，上游项将被丢弃，只保留终端事件。终止的 DirectProcessor 将向迟到的订阅者发出终端信号。