从多个供应商生成一个流答案

【问题标题】：Generate one Stream from multiple Suppliers从多个供应商生成一个流
【发布时间】：2014-08-14 15:11:05
【问题描述】：

我正在尝试使用 Java 8 Lambda/Stream API 来建模一个简单的生产者/消费者系统，就像：

    Stream<Sample> samplesFlow = Stream.generate(new SampleSupplier());
    samplesFlow.forEach(new SampleConsumer());

我意识到向多个消费者“扩展”非常简单：

    samplesFlow
        .peek(new SampleConsumer1())
        .peek(new SampleConsumer2())
        .forEach(new SampleConsumer3());

但是向系统添加新的生产者呢？是否有一些惯用或“优雅”的方式来从多个无限供应商生成流？喜欢：

Stream.generate(new SampleSupplier1(),new SampleSupplier2());  // made it up

每个供应商都模拟一个从远程源获取数据的网络侦听器。

提前致谢！

【问题讨论】：

【解决方案1】：

您没有指定要如何组合提供的值。

如果你想让这些值交替出现，一个解决方案是：

Stream.generate(supplier1).flatMap(x->Stream.of(x, supplier2.get()))

如果你想拥有某种配对，你可以使用

Stream.generate(()->new Pair<>(supplier1.get(), supplier2.get()))

尽管由您来编写Pair 类，因为 jdk 不提供这样的类。（你可以滥用 AbstractMap.SimpleEntry 但这很讨厌）。

如果您有一个有限的Stream，您可以使用Stream.concat 创建一个流，该流将在第二个项目之前处理第一个流的所有项目，但是，使用Supplier 创建的流默认情况下是无限的所以你必须在第一个流上使用limit，然后才能将它连接到另一个流，所以这不是一个通用的解决方案。

如果你想对每个Supplier只查询一次，你可以使用

Stream.of(supplier1, supplier2).map(Supplier::get)

当然，如果您不需要对 Suppliers 进行惰性求值，那么

Stream.of(supplier1.get(), supplier2.get())

也可以。

【讨论】：

我想也许 mananoreboton 想按顺序流式传输多个 Streams 或 Suppliers。这可以通过Stream.concat 来完成。
@VGR：这不起作用，因为从Supplier 生成的流是无限的，因此您不能一个接一个地处理一个流。我会扩展我的答案，因为这似乎一遍又一遍地弹出。
非常好。我不知道为什么这对我来说不是很明显……
Supplier 每次调用时都会提供一个值。通过反复调用Supplier，generate 使其无限。 OP 可能希望使用 n Supplier 实例来生成 n 值的有限流。我会问的。
@Stuart Marks：你是对的，这个问题有点不明确。然而，对我来说，Supplier 在概念上是一个无限源，因为它不要求多久调用一次get，并且没有像hasNext 方法这样的东西（不像Iterator）。但我添加了一次一次Supplier 替代方案。

【解决方案2】：

为了完整起见，另一个隐藏实现复杂性的选项是使用

StreamSupport.stream(Spliterator<T> spliterator, boolean parallel)

生成流并将Spliterator 委托给遍历不同来源的元素。

在实现forEachRemaining(Consumer<? super T> action)时需要牢记一些注意事项

【讨论】：