别忘了给我一个 .connect 离子对每个 ConnectableFlux

在所有这些例子中,都遗漏了 .连接 ion声明。

对于第一种情况,要使其发挥作用,我们必须 .连接 到第一个 publish 可连接通量 也:

ConnectableFlux<Integer> publish = Flux.just(1)
        .publish();

ConnectableFlux<Integer> publish1 = Flux.just(2)
        .flatMap(x -> publish)
        .publish();

publish1.subscribe(System.out::println, System.out::println, System.out::println);
publish1.connect();
publish.connect();

对于下面的两个例子,我们有一些相似的东西。当我们使用 Flux.just(...).publish().publish() 我们创建了两个 可连接通量 . 这里的问题是第一个被删除了。如果必须进行后续 .publish ing(这是非常不合逻辑的)我们可以使用以下技术来避免删除以前的 可连接通量 锿:

ConnectableFlux<Integer> publish1 = Flux.just(2)
        .publish()
        .autoConnect() // or .autoConnect(0)
        .doOnNext(System.out::println)
        .publish();

publish1.subscribe(System.out::println, System.out::println, System.out::println);
publish1.connect();

在该示例中,我们使用 .autoConnect() 操作员,如果 .autoConnect(0) 只是 ConnectableFlux#connect 和 return this; 声明。万一 .autoConnect(>0) 用于对初始源的延迟订阅,听起来像“ 仅当我们得到 N 订阅者 "