在Java 8中将两个流相加，或向流中添加额外的元素

Question

我可以添加流或额外的元素，就像这样：

Stream stream = Stream.concat(stream1, Stream.concat(stream2, Stream.of(element));

我可以随时添加新内容，比如这个：

Stream stream = Stream.concat(
                       Stream.concat(
                              stream1.filter(x -> x!=0), stream2)
                              .filter(x -> x!=1),
                                  Stream.of(element))
                                  .filter(x -> x!=2);

但这很丑陋，因为concat是静态的。如果concat是一个实例方法，上面的例子将会更容易阅读：

 Stream stream = stream1.concat(stream2).concat(element);

而

 Stream stream = stream1
                 .filter(x -> x!=0)
                 .concat(stream2)
                 .filter(x -> x!=1)
                 .concat(element)
                 .filter(x -> x!=2);

我的问题是：

1）concat为什么是静态的？是否有相应的实例方法我没有发现？

2）无论如何，有更好的方法来实现这个吗？

- MarcG

---

看起来事情并不总是这样，但我只是找不到原因。 - Edwin Dalorzo

Answer 1

很不幸，这个答案可能没有什么帮助，但我曾对Java Lambda邮件列表进行了取证分析，以查找这种设计的原因。以下是我的发现。

最初存在Stream.concat（Stream）的实例方法

在邮件列表中，我可以清楚地看到该方法最初是作为实例方法实现的，正如您可以在Paul Sandoz在此线程中所阅读的关于concat操作的内容。

他们讨论了在流可能无限时可能出现的问题以及在这些情况下连接意味着什么，但我不认为这是修改的原因。

您可以在此其他线程中看到，JDK 8的一些早期用户质疑使用空参数时concat实例方法的行为。

然而，此其他线程揭示了concat方法的设计正在讨论中。

重构为Streams.concat（Stream，Stream）

但是突然之间，这些方法被更改为静态方法，没有任何解释，正如您可以在此关于组合流的线程中看到的那样。这也许是唯一一个关于这种更改的邮件线程，但它对我来说并不足够清楚，无法确定重构的原因。但我们可以看到他们进行了提交，建议将concat方法从Stream移动到助手类Streams中。

重构为Stream.concat（Stream，Stream）

稍后，它又被移回到Stream，但同样没有解释。

总之，对于我来说，设计的原因还不是完全清楚，我找不到一个好的解释。我想你仍然可以在邮件列表中提出问题。

流连接的一些替代方法

Michael Hixson在此其他线程中讨论/询问了结合/连接流的其他方法。

1. 要组合两个流，应该这样做：

   Stream.concat(s1, s2)
   

   不是这个：

   Stream.of(s1, s2).flatMap(x -> x)
   

   ...对吗？

2. 要合并两个以上的流，我应该这样做：

Stream.of(s1, s2, s3, ...).flatMap(x -> x)

不是这个：

Stream.of(s1, s2, s3, ...).reduce(Stream.empty(), Stream::concat)

...对吧？

- Edwin Dalorzo

---

不错的研究。我将使用这个带有可变参数的 Stream.concat 方法：public static <T> Stream<T> concat(Stream<T>... streams) { return Stream.of(streams).reduce(Stream.empty(), Stream::concat);} - MarcG

今天我写了自己的concat版本，然后就发现了这个主题。签名略有不同，但多亏了它更通用;)例如，您可以将Stream<Integer>和Stream<Double>合并为Stream<Number>。@SafeVarargs private static <T> Stream<T> concat(Stream<? extends T>... streams) { return Stream.of(streams).reduce(Stream.empty(),Stream::concat).map(Function.identity());} - kant

你尝试在IDE中输入了吗？如果没有使用.map(identity())，你将会得到编译错误。我想返回Stream<T>，但是语句：return Stream.of(streams).reduce(Stream.empty(),Stream::concat) 返回的是Stream<? extends T>。（Something<T>是Something<? extends T>的子类型，而不是反过来，所以不能强制转换）。额外的.map(identity())将<? extends T>转换为<T>。这是由于Java 8方法参数和返回类型的“目标类型”以及map()方法的签名混合导致的。实际上，它是Function.<T>identity()。 - kant

@kant，我刚刚顺利地完成了。我猜你的问题可能出在泛型上，而不是流连接惯用语上。 - Edwin Dalorzo

@kant，我认为使用? extends T没有太多意义，因为你可以使用捕获转换。无论如何，这是我的代码片段，我们在Gist中继续讨论。 - Edwin Dalorzo

Answer 2

如果您为Stream.concat和Stream.of添加静态导入，则第一个示例可以编写如下：

Stream<Foo> stream = concat(stream1, concat(stream2, of(element)));

导入具有通用名称的静态方法可能会导致代码难以阅读和维护（命名空间污染）。因此，最好创建自己的静态方法并赋予更有意义的名称。但是，为了演示，我将继续使用这个名称。

public static <T> Stream<T> concat(Stream<? extends T> lhs, Stream<? extends T> rhs) {
    return Stream.concat(lhs, rhs);
}
public static <T> Stream<T> concat(Stream<? extends T> lhs, T rhs) {
    return Stream.concat(lhs, Stream.of(rhs));
}

使用这两个静态方法（可选与静态导入结合使用），可以将这两个示例编写为以下形式：

Stream<Foo> stream = concat(stream1, concat(stream2, element));

Stream<Foo> stream = concat(
                         concat(stream1.filter(x -> x!=0), stream2).filter(x -> x!=1),
                         element)
                     .filter(x -> x!=2);

代码现在更加简短了。但是，我同意可读性并没有得到提高。所以我有另一个解决方案。

---

在许多情况下，Collectors可以用于扩展流的功能。通过下面的两个Collectors，可以将两个示例编写如下：

Stream<Foo> stream = stream1.collect(concat(stream2)).collect(concat(element));

Stream<Foo> stream = stream1
                     .filter(x -> x!=0)
                     .collect(concat(stream2))
                     .filter(x -> x!=1)
                     .collect(concat(element))
                     .filter(x -> x!=2);

你所期望的语法和上面的语法唯一的区别是，你需要用 collect(concat(...)) 替换 concat(...)。这两个静态方法可以按照以下方式实现（可选与静态导入一起使用）：

private static <T,A,R,S> Collector<T,?,S> combine(Collector<T,A,R> collector, Function<? super R, ? extends S> function) {
    return Collector.of(
        collector.supplier(),
        collector.accumulator(),
        collector.combiner(),
        collector.finisher().andThen(function));
}
public static <T> Collector<T,?,Stream<T>> concat(Stream<? extends T> other) {
    return combine(Collectors.toList(),
        list -> Stream.concat(list.stream(), other));
}
public static <T> Collector<T,?,Stream<T>> concat(T element) {
    return concat(Stream.of(element));
}

当然，这种解决方案有一个需要提及的缺点。collect是一种消耗流中所有元素的终止操作。此外，每次在链中使用collector concat都会创建一个中间ArrayList。这两个操作都可能对程序的行为产生重大影响。然而，如果可读性比性能更重要，这仍然可能是一种非常有用的方法。

- nosid

---

жҲ‘и®Өдёәconcat收йӣҶеҷЁдёҚеӨӘжҳ“иҜ»гҖӮиҝҷдёӘеҚ•еҸӮж•°йқҷжҖҒж–№жі•зҡ„е‘ҪеҗҚж–№ејҸеҫҲеҘҮжҖӘпјҢиҖҢдё”дҪҝз”ЁcollectжқҘиҝӣиЎҢиҝһжҺҘд№ҹеҫҲеҘҮжҖӘгҖӮ - Didier L

@nosid，或许这个问题与本帖有些不相关，但是你为什么认为“导入具有名称的静态方法是一个坏主意”？我真的很感兴趣——我发现它使代码更简洁易读，而且我问过的很多人也持相同看法。能否提供一些例子说明通常为什么不好呢？ - quantum

@Quantum：compare(reverse(getType(42)), of(6 * 9).hashCode())的意思是什么？请注意，我并没有说静态导入是一个坏主意，但是像of和concat这样的通用名称的静态导入是不好的。 - nosid

@nosid：在现代IDE中悬停在每个语句上不会很快地显示其含义吗？无论如何，我认为这最多只能算是个人偏好的陈述，因为我仍然看不到静态导入“通用”名称存在任何技术上的问题 - 除非你正在使用记事本或VI（M）进行编程，在这种情况下你有更大的问题。 - quantum

我不打算说Scala SDK更好，但是…哎呀，我说了。 - eirirlar

Answer 3

只需做：

Stream.of(stream1, stream2, Stream.of(element)).flatMap(identity());

其中identity()是Function.identity()的静态导入。

将多个流连接成一个流与展开流相同。

然而，不幸的是，由于某种原因，在Stream上没有flatten()方法，所以您必须使用带有身份函数的flatMap()。

- herman

Answer 4

我的StreamEx库扩展了Stream API的功能。具体而言，它提供了像append和prepend这样的方法来解决此问题（内部它们使用concat）。这些方法可以接受另一个流、集合或可变参数数组。使用我的库，您可以通过以下方式解决问题（请注意，对于非原始流，x != 0看起来很奇怪）：

Stream<Integer> stream = StreamEx.of(stream1)
             .filter(x -> !x.equals(0))
             .append(stream2)
             .filter(x -> !x.equals(1))
             .append(element)
             .filter(x -> !x.equals(2));

顺便提一下，对于你的filter 操作也有一个快捷方式：

Stream<Integer> stream = StreamEx.of(stream1).without(0)
                                 .append(stream2).without(1)
                                 .append(element).without(2);

- Tagir Valeev

Answer 5

你可以使用Guava的Streams.concat(Stream<? extends T>... streams)方法，这将产生一个扁平化的流：

Stream stream = Streams.concat(stream1, stream2, Stream.of(element));

- Kunda

---

Stream.of 不应该被静态导入。它的读法是“stream of”，类似于 Optional.of 的读法是“optional of”。你甚至不能在同一个文件中同时导入 Stream.of 和 Optional.of。 - herman

@herman 你说得对 - 我修改了我的回答，侧重于Guava带来的简洁性，并将静态导入的使用（或滥用）留给读者自行决定。 - Kunda

Answer 6

如果你不介意使用第三方库，cyclops-react 有一个扩展的Stream类型，可以通过追加/预置运算符来实现。可以使用实例方法追加和预置单个值、数组、可迭代对象、流或反应式流发布者。

Stream stream = ReactiveSeq.of(1,2)
                           .filter(x -> x!=0)
                           .append(ReactiveSeq.of(3,4))
                           .filter(x -> x!=1)
                           .append(5)
                           .filter(x -> x!=2);

[披露：我是cyclops-react的首席开发人员]

- John McClean

Answer 7

你考虑编写自己的 concat 方法如何？

public static <T> Stream<T> concat(Stream<? extends T> a, 
                                   Stream<? extends T> b, 
                                   Stream<? extends T>... args)
{
    Stream<T> concatenated = Stream.concat(a, b);
    for (Stream<? extends T> stream : args)
    {
        concatenated = Stream.concat(concatenated, stream);
    }
    return concatenated;
}

这样至少可以让你的第一个例子更易读。

正如@Legna所指出的那样，由于Stream::concat的嵌套调用，这可能很快导致StackOverflowError。

因此，这里提供另一种版本，应该可以解决问题并且看起来相当整洁：

public static <T> Stream<T> concat(final Stream<? extends T>... args)
{
    return args == null ? Stream.empty()
                        : Stream.of(args).flatMap(Function.identity());
}

- Felix S

---

在构建重复连接的流时要小心。访问深度连接流的元素可能会导致深度调用链，甚至是 StackOverflowError。 - Legna

Answer 8

说到底，我不关心合并流，而是获得处理所有这些流中每个元素的合并结果。

虽然合并流可能会证明很麻烦（因此需要这个主题），但合并它们的处理结果相当容易。

解决的关键是创建自己的收集器，并确保新收集器的供应函数每次都返回相同的集合（而不是一个新的集合），下面的代码说明了这种方法。

package scratchpad;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collector;
import java.util.stream.Stream;

public class CombineStreams {
    public CombineStreams() {
        super();
    }

    public static void main(String[] args) {
        List<String> resultList = new ArrayList<>();
        Collector<String, List<String>, List<String>> collector = Collector.of(
                () -> resultList,
                (list, item) -> {
                    list.add(item);
                },
                (llist, rlist) -> {
                    llist.addAll(rlist);
                    return llist;
                }
        );
        String searchString = "Wil";

        System.out.println("After processing first stream\n"
                + createFirstStream().filter(name -> name.contains(searchString)).collect(collector));
        System.out.println();

        System.out.println("After processing second stream\n"
                + createSecondStream().filter(name -> name.contains(searchString)).collect(collector));
        System.out.println();

        System.out.println("After processing third stream\n"
                + createThirdStream().filter(name -> name.contains(searchString)).collect(collector));
        System.out.println();

    }

    private static Stream<String> createFirstStream() {
        return Arrays.asList(
                "William Shakespeare",
                "Emily Dickinson",
                "H. P. Lovecraft",
                "Arthur Conan Doyle",
                "Leo Tolstoy",
                "Edgar Allan Poe",
                "Robert Ervin Howard",
                "Rabindranath Tagore",
                "Rudyard Kipling",
                "Seneca",
                "John Donne",
                "Sarah Williams",
                "Oscar Wilde",
                "Catullus",
                "Alfred Tennyson",
                "William Blake",
                "Charles Dickens",
                "John Keats",
                "Theodor Herzl"
        ).stream();
    }

    private static Stream<String> createSecondStream() {
        return Arrays.asList(
                "Percy Bysshe Shelley",
                "Ernest Hemingway",
                "Barack Obama",
                "Anton Chekhov",
                "Henry Wadsworth Longfellow",
                "Arthur Schopenhauer",
                "Jacob De Haas",
                "George Gordon Byron",
                "Jack London",
                "Robert Frost",
                "Abraham Lincoln",
                "O. Henry",
                "Ovid",
                "Robert Louis Stevenson",
                "John Masefield",
                "James Joyce",
                "Clark Ashton Smith",
                "Aristotle",
                "William Wordsworth",
                "Jane Austen"
        ).stream();
    }

    private static Stream<String> createThirdStream() {
        return Arrays.asList(
                "Niccolò Machiavelli",
                "Lewis Carroll",
                "Robert Burns",
                "Edgar Rice Burroughs",
                "Plato",
                "John Milton",
                "Ralph Waldo Emerson",
                "Margaret Thatcher",
                "Sylvie d'Avigdor",
                "Marcus Tullius Cicero",
                "Banjo Paterson",
                "Woodrow Wilson",
                "Walt Whitman",
                "Theodore Roosevelt",
                "Agatha Christie",
                "Ambrose Bierce",
                "Nikola Tesla",
                "Franz Kafka"
        ).stream();
    }
}

- Legna