Mono.count 无限期阻塞

Question

我运行了这个：

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

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

它会产生：

mono.block: null    

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

另一个例子，类似于现实世界的场景。我有一个源通量，例如：

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() 无限期阻塞。

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

Answer 1

您可以使用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% 确定它是正确的。

1. .publish() 将上游源变成热流
2. 然后您订阅了两次（但这些还没有启动流程，因为可连接的通量尚未连接到上游）
3. .connect() 订阅上游，并启动流
4. 上游，以及在connect() 之前注册的两个订阅完成（因为这一切都发生在主线程中）
5. 此时，ConnectableFlux 不再连接到上游，因为上游已完成（反应器文档对在上游源完成后新订阅到达时 ConnectableFlux 发生的情况的详细信息，所以这就是我我不是 100% 确定的。）
6. block() 创建一个新的订阅。
7. 但由于 ConnectableFlux 不再连接，因此没有数据流动
8. 如果您再次调用connect()（从另一个线程，因为主线程被阻塞），数据将再次流动，block() 将完成。但是，这将是一个新序列（不是在步骤 4 中完成的原始序列）

为什么我的示例有效？

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