java - 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 的预期行为。我刚刚读完 documentation并发现以下有关 DirectProcessor 的内容:

Once the Processor has terminated (usually through its sink’s error(Throwable) or complete() methods being called), it lets more subscribers subscribe but replays the termination signal to them immediately.

因此，由于 repeat 只是重新订阅，因此将立即对它们调用 onComplete 处理程序。您确定需要DirectProcessor吗？

编辑:此行为也记录在案 here

Note: If there are no Subscribers, upstream items are dropped and only the terminal events are retained. A terminated DirectProcessor will emit the terminal signal to late subscribers.