为什么 Thread.sleep() 会触发订阅 Flux.interval()？答案

【问题标题】：Why does Thread.sleep() trigger the subscription to Flux.interval()?为什么 Thread.sleep() 会触发订阅 Flux.interval()？
【发布时间】：2020-05-13 05:58:11
【问题描述】：

如果，在main() 方法中，我执行这个

Flux.just(1,2)
    .log()
    .subscribe();

我在控制台中得到了这个：

[ INFO] (main) | onSubscribe([Synchronous Fuseable] FluxArray.ArraySubscription)
[ INFO] (main) | request(unbounded)
[ INFO] (main) | onNext(1)
[ INFO] (main) | onNext(2)
[ INFO] (main) | onComplete()

如果我使用interval() 方法而不是just()：

Flux.interval(Duration.ofMillis(100))
    .take(2)
    .log()
    .subscribe();

元素不会被记录，除非我添加 Thread.sleep() 这给了我：

[ INFO] (main) onSubscribe(FluxTake.TakeSubscriber)
[ INFO] (main) request(unbounded)
[ INFO] (parallel-1) onNext(0)
[ INFO] (parallel-1) onNext(1)
[ INFO] (parallel-1) onComplete()

问题是：为什么我需要暂停线程才能真正触发订阅？

【问题讨论】：

标签： project-reactor

【解决方案1】：

您需要等待主线程并让执行完成。您的主线程在生成下一个元素之前终止。您的第一个元素是在100 ms 之后生成的，因此您需要等待/阻塞主线程。试试这个：

CountDownLatch latch = new CountDownLatch(1);
Flux.interval(Duration.ofMillis(100))
        .take(2)
        .doOnComplete(latch::countDown)
        .log()
        .subscribe();
latch.await(); // wait for doOnComplete

CountDownLatch:

一种同步辅助工具，允许一个或多个线程等待，直到在其他线程中执行的一组操作完成。

【讨论】：

为什么不直接使用 .blockLast()。这是做同样事情的复杂方式。
@PrashantPandey 如果有订阅者怎么办？ blockLast 没有意义。
如果你改用 blocklast 会发生什么？
@TobiAkinyemi 这样就不需要等待闩锁了，但这违背了订阅者的目的，因为它是一个。
@AniketSahrawat 不是非阻塞订阅吗？如果你使用blocklast，你就不需要latch

【解决方案2】：

Flux.interval(..) 在parallel 上发出项目，从而释放您的调用线程。这就是您的程序退出的原因。你应该做的是：

Flux.interval(Duration.of(duration))
    .take(n)
    .doOnNext(this::logElement)
    .blockLast();

这将阻塞调用线程，直到doOnNext() 最后一次发出（应该是上游的第二个项目）。

【讨论】：

我错过了他正在订阅链。如果有订阅，此解决方案不起作用。

【解决方案3】：

Flux.interval(Duration) 产生一个无限的 Flux 并从时钟发出有规律的滴答声。

所以你的第一个例子没有并发，一切都发生在同一个线程上。订阅和所有事件必须在方法和流程结束之前完成。

通过使用 Flux.interval，您在示例中添加了并发和异步行为。间隔是像节拍器一样的常规时钟。您订阅但立即完成，并且该过程通常会在任何工作（onNext）完成之前完成。

【讨论】：

在第二个例子中，如果我不暂停 main 进程，它会立即停止，但是进程 parallel-1 会发生什么？在控制台中没有记录任何内容时它是否仍在运行？
测试一下。我会假设 JVM 在您的测试中退出。在服务器中，它可能最终完成。但这不是很好的代码。在不保留结果和管理生命周期的情况下创建订阅可能会导致负载不足。