如果创建热发布者，flux cache()、replay() 和 publish() 有什么区别？答案

【问题标题】：What is the difference between flux cache(), replay() and publish() if creating a hot publisher?如果创建热发布者，flux cache()、replay() 和 publish() 有什么区别？
【发布时间】：2021-09-28 16:20:01
【问题描述】：

flux cache()、replay() 和 publish() 创建热门发布者有什么区别？对于哪个用例，哪个运算符最适合？

以下示例重放了 3 种不同方法的所有 5 个元素。

cache():

        var flux = Flux.fromStream(Stream.of(1,2,3,4,5))
                .delayElements(Duration.ofSeconds(1)).cache();

        flux.doOnNext(v -> System.out.println("First: " + v))
        .subscribe();

        Thread.sleep(5000);

        flux.doOnNext(v -> System.out.println("Second: " + v))
                .subscribe();

        Thread.sleep(10000);

replay():

        var flux = Flux.fromStream(Stream.of(1,2,3,4,5))
                .delayElements(Duration.ofSeconds(1)).replay();

        flux.doOnNext(v -> System.out.println("First: " + v))
        .subscribe();

        Thread.sleep(5000);

        flux.doOnNext(v -> System.out.println("Second: " + v))
                .subscribe();

        flux.connect();

        Thread.sleep(10000);

publish():

        var flux = Flux.fromStream(Stream.of(1,2,3,4,5))
                .delayElements(Duration.ofSeconds(1)).publish();

        flux.doOnNext(v -> System.out.println("First: " + v))
        .subscribe();

        Thread.sleep(5000);

        flux.doOnNext(v -> System.out.println("Second: " + v))
                .subscribe();

        flux.connect();

        Thread.sleep(10000);

打印结果的一种变体：

First: 1
First: 2
First: 3
First: 4
Second: 1
Second: 2
Second: 3
Second: 4
First: 5
Second: 5

【问题讨论】：

标签： spring-webflux project-reactor

【解决方案1】：

cache() 是.replay().autoConnect(1) 的方便别名，即。只要第一个订阅者进来，它就会为您执行connect()。

但由于它会重放整个历史记录，因此第二个订阅者仍然可以看到所有元素。

从您的replay() 和publish() 示例中，您可能认为两者之间没有区别。但那是因为你connect() AFTER 两个订阅者都订阅了...

如果您将connect() 呼叫移动到第二个订阅者之前，您会看到在publish() 的情况下它看不到任何值。另一方面，replay() 会将整个历史重播给第二个订阅者，尽管它已经晚了。

【讨论】：