project-reactor - Flux.create 和 Flux.generate 的区别

Question

Flux.create 和有什么区别和 Flux.generate ?我正在寻找 - 最好是使用示例用例 - 了解我何时应该使用其中一个。

Answer 1

简而言之:

Flux::create doesn't react to changes in the state of the app while Flux::generate does.

长版 Flux::create当您想要计算不受应用程序状态和管道状态影响的多个 (0...infinity) 值时，您将使用它(您的管道 == Flux::create == 下游的操作链之后) )。
为什么？因为您发送到 Flux::create 的方法一直在计算元素(或没有)。下游将决定它想要多少个元素(元素 == 下一个信号)，如果他跟不上，那些已经发出的元素将在某种策略中被删除/缓冲(默认情况下，它们将被缓冲，直到下游将索要更多)。
第一个也是最简单的用例是发出值，理论上，您可以将这些值总结为一个集合，然后才获取每个元素并对其进行处理:

Flux<String> articlesFlux = Flux.create((FluxSink<String> sink) -> {
/* get all the latest article from a server and emit them one by one to downstream. */
List<String> articals = getArticalsFromServer();
articals.forEach(sink::next);
});

如您所见， Flux.create 用于阻塞方法( getArticalsFromServer )与异步代码之间的交互。
我确定 Flux.create 还有其他用例。
Flux::generate

Flux.generate((SynchronousSink<Integer> synchronousSink) -> {
        synchronousSink.next(1);
    })
    .doOnNext(number -> System.out.println(number))
    .doOnNext(number -> System.out.println(number + 4))
    .subscribe();

输出将为 1 5 1 5 1 5................forever 在您发送到 Flux::generate 的方法的每次调用 中， synchronousSink 只能发出: onSubscribe onNext? (onError | onComplete)? 。
这意味着 Flux::generate 将 计算并发出 值 按需 。你应该什么时候使用它？在计算可能不会在下游使用的元素或您发出的事件受应用程序状态或管道状态影响的情况下成本太高(您的管道 == Flux::create 之后的操作链 = = 下游)。
例如，如果您正在构建一个 Torrent 应用程序，那么您将实时接收数据块。您可以使用 Flux::generate 将任务(要下载的块)分配给多个线程，并且只有在某个线程询问时，您才会在 Flux::generate 中计算要下载的块。所以你只会发出你没有的块。与 Flux::create 相同的算法将失败，因为 Flux::create 将发出我们没有的所有块，如果某些块无法下载，那么我们就有问题了。因为 Flux::create 不会对应用程序状态的变化使用react，而 Flux::generate 会。