反应器：Flux<object> .subscribe() 与 .toStream()

Question

我有一个函数：Flux queryPerson()，它查询数据库以生成对象并在 Flux 中返回它们。当我使用 .subscribe() 时，应用程序只是通过代码运行并退出。它不会等待结果返回查询。但是当我使用 .toStream() 阻止流时，我可以看到打印输出。我做错了什么？

personRepository
    .queryPerson()
    .map(x -> x.getFirst().concat("ok"))
    .subscribe(i -> System.out.println(i))
    //.toStream().forEach(System.out::println)
;

Answer 1

我假设您没有某种网络应用程序，而是一个命令行运行器或简单的 Java 应用程序。考虑到应用程序在异步任务之前完成是正常的。

.订阅

订阅是一种使用传入数据的异步方式，在您订阅 Flux 后，您会立即将控制权返回给调用线程。

这正是反应式编程的工作方式，您定义行为，您可以在其他线程和调用线程中以很好的抽象方式运行它。

正如它在Flux docs 中所说的那样

由于序列可以异步，这将立即返回 控制到调用线程。这可以给人的印象 在主线程或单元测试中执行时不会调用使用者 例如。

.toStream

另一方面，使用 .toStream 您会收到一个 Java 流，即使它的大小未知，您仍然可以像普通 Java 流一样以同步方式对其进行迭代。

更多解释可见.toStream docs of Flux