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

Question

我有一个函数：Flux queryPerson()，它查询数据库以生成对象并 return 它们在 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 后，您会立即 return 控制调用线程。

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

如 Flux docs

中所述

since the sequence can be asynchronous, this will immediately return control to the calling thread. This can give the impression the consumer is not invoked when executing in a main thread or a unit test for instance.

.toStream

另一方面，使用 .toStream 你会收到一个 Java 流，即使它的大小未知，你仍然可以像正常 Java 流。

更多解释可以在.toStream docs of Flux

中找到

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

Reactor: Flux<object> .subscribe() vs. .toStream()

java

reactor

flux

spring-reactive

.订阅

.toStream