Infinite Java Stream 和 Reactor Flux 的区别答案

【问题标题】：Difference between Infinite Java Stream and Reactor FluxInfinite Java Stream 和 Reactor Flux 的区别
【发布时间】：2019-03-20 02:06:31
【问题描述】：

我正在尝试分别找出无限流和无限通量之间的概念差异（如果有的话）。

就此而言，我提出了以下无限流/通量的示例

@Test
public void infinteStream() {

  //Prints infinite number of integers
  Stream<Integer> infiniteStream = Stream.iterate(0, i -> i+1);
 
  infiniteStream.forEach(System.out::println);
}

@Test
public void infiniteFlux()  {
    
   //Prints infinite number of date strings (every second)
   Flux<LocalDateTime> localDateTimeFlux = Flux.interval(Duration.ofSeconds(1))
            .map(t -> LocalDateTime.now());

    localDateTimeFlux.subscribe(t -> System.out.println(t.format(DateTimeFormatter.ofPattern("yyyy-MM-dd hh:mm:ss"))));
}

关于这些例子，我有一个问题：infinteStream() 是否有 Flux 的类比（以及 infinteFlux() 和 Stream 的类比）？而且，更一般地说，无限流和通量之间有什么区别吗？

【问题讨论】：

标签： java java-stream flux spring-webflux reactor

【解决方案1】：

Stream 和 Flux 完全不同：

Stream 是单次使用，而您可以多次订阅Flux
Stream 是基于拉动的（消耗一个元素调用下一个元素），而Flux 具有混合推/拉模型，其中发布者可以推送元素但仍必须尊重消费者发出的背压信号
Stream 是同步序列，Flux 可以表示异步序列

在示例中，您使用Stream 生成无限的值序列，它们会以尽可能快的速度生成和消耗。在您的 Flux 示例中，您正在以固定的时间间隔生成值（我不确定您是否可以使用 Stream 执行此操作）。使用Flux，您还可以对Flux.generate 进行无间隔序列，就像您的Stream 示例一样。

一般而言，您可以将Flux 视为Stream + CompletableFuture 的组合，其中：

很多强大的运算符
背压支持
控制发布者和订阅者的行为
控制时间概念（缓冲值窗口、添加超时和回退等）
为通过网络（从数据库或远程 Web API）获取的异步序列量身定制的东西

【讨论】：

亲爱的布赖恩，感谢您阐明差异！我会用 Flux.generate 试试你的建议 :-)
在什么情况下应该使用流而不是通量？

【解决方案2】：

作为参考，与此同时，我为infiniteFlux() 提出了一个流解决方案：

@Test 
public void infiniteFluxWithStream()  {

    Stream<Integer> infiniteStream = Stream.iterate(0, i -> i+1).peek(x->{
    LocalDateTime t = LocalDateTime.now();
    t.format(DateTimeFormatter.ofPattern("yyyy-MM-dd hh:mm:ss"));
    System.out.println(t);
    });

    infiniteStream.forEach(x->{
    try {
        Thread.sleep(10000);
    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    });

}

这确实很难看。但是，它表明在（非常）原则上，可以根据 Streams 重写简单的 Flux-Examples。

【讨论】：