java - 无限计数块

我跑了这个：

Mono<Void> mono = Mono.empty();

System.out.println("mono.block: " + mono.block());

它产生：

mono.block: null

如预期的那样。换句话说，如果block已经完成，则调用Mono将立即返回。

另一个例子，类似于真实场景。我有一个源流量，例如：

Flux<Integer> ints = Flux.range(0, 2);

我创建了一个可连接的流量，用于允许多个订阅者：

ConnectableFlux<Integer> publish = ints.publish();

对于此示例，假设有一个实际的订户：

publish
   .doOnComplete(() -> System.out.println("publish completed"))
   .subscribe();

另一个只产生元素计数的订阅者：

Mono<Long> countMono = publish
   .doOnComplete(() -> System.out.println("countMono completed"))
   .count();

countMono.subscribe();

我连接可连接的磁通并打印元素计数：

publish.connect();

System.out.println("block");

long count = countMono.block();

System.out.println("count: " + count);

打印：

publish completed
countMono completed
block

换句话说，两个订户都成功完成订阅，但随后countMono.block()无限期阻塞。

为什么会这样，我该如何做呢？我的最终目标是获取元素的数量。

最佳答案

您可以使用autoConnect或refCount代替手动调用connect()使其工作。

例如：

        Flux<Integer> ints = Flux.range(0, 2);
        Flux<Integer> publish = ints.publish()
                .autoConnect(2);  // new
        publish
                .doOnComplete(() -> System.out.println("publish completed"))
                .subscribe();
        Mono<Long> countMono = publish
                .doOnComplete(() -> System.out.println("countMono completed"))
                .count();
        // countMono.subscribe();
        long count = countMono.block();
        System.out.println("count: " + count);

为什么您的示例不起作用？

这是我认为您的示例中正在发生的事情……但这是基于我的有限知识，我不确定100％是否正确。


.publish()将上游源变成热门流
然后，您需要订阅两次（但是这些还没有启动流程，因为可连接的流量尚未连接到上游）
.connect()订阅上游，并开始流程
上游，以及在connect()之前完成注册的两个预订（因为这全部发生在主线程中）
此时，ConnectableFlux不再连接到上游，因为上游已经完成（反应堆文档详细介绍了在上游源完成后到达新订阅时ConnectableFlux发生了什么情况，所以我不是这样。 100％可以确定。）
block()创建一个新的订阅。
但是由于ConnectableFlux不再连接，所以没有数据在流动
如果要再次调用connect()（从另一个线程，因为主线程被阻塞），数据将再次流动，并且block()将完成。但是，这将是一个新序列（不是在步骤4中完成的原始序列）


为什么我的示例有效？

仅创建两个预订（而不是示例中的3个），一个来自.subscribe()调用，一个来自.block()。 2个订阅后，ConnectableFlux自动连接，因此block()订阅完成。两个订阅共享相同的上游序列。