Java 8 Stream API 教程

1. 概述

在这个综合教程中，我们将介绍 Java 8 Streams 从创建到并行执行的实际用途。

要理解本材料，读者需要具备 Java 8（lambda 表达式、可Optional、方法引用）和 Stream API 的基本知识。为了更熟悉这些主题，请查看我们之前的文章：Java 8 中的新功能 和 Java 8 Streams 简介。

2. 创建流

有多种方法可以创建不同来源的流实例。创建后，实例将不会修改其源，因此允许从单个源创建多个实例。

2.1.空流

如果创建空流，我们应该使用 empty() 方法：

Stream<String> streamEmpty = Stream.empty();

我们经常在创建时使用 empty() 方法，以避免在没有元素的流中返回 null：

public Stream<String> streamOf(List<String> list) {
    return list == null || list.isEmpty() ? Stream.empty() : list.stream();
}

2.2. 集合流

我们还可以创建任何类型的集合（Collection、List、Set）流：

Collection<String> collection = Arrays.asList("a", "b", "c");
Stream<String> streamOfCollection = collection.stream();

2.3.数组流

数组也可以是流的源：

Stream<String> streamOfArray = Stream.of("a", "b", "c");

我们还可以从现有数组或数组的一部分创建流：

String[] arr = new String[]{"a", "b", "c"};
Stream<String> streamOfArrayFull = Arrays.stream(arr);
Stream<String> streamOfArrayPart = Arrays.stream(arr, 1, 3);

2.4.Stream.builder()

使用 builder 时，应在语句的右侧部分额外指定所需的类型， 否则 build() 方法将创建一个 Stream<Object>:实例：

Stream<String> streamBuilder =
  Stream.<String>builder().add("a").add("b").add("c").build();

2.5. Stream.generate()

The generate() 方法接受Supplier<T> 来生成元素。由于生成的数据流是无限的，开发者应指定所需的大小，否则 generate() 方法将一直工作到达到内存限制为止：

Stream<String> streamGenerated =
  Stream.generate(() -> "element").limit(10);

上面的代码创建了一个由十个字符串组成的序列，其值为“element.”

2.6. Stream.iterate()

创建无限流的另一种方法是使用 iterate() 方法：

Stream<Integer> streamIterated = Stream.iterate(40, n -> n + 2).limit(20);

结果流的第一个元素是 iterate() 方法的第一个参数。创建每个后续元素时，指定的函数将应用于前一个元素。在上面的示例中，第二个元素将为 42。

2.7.Primitive流

Java 8 提供了从三种基本类型创建流的可能性： int, long 和 double。 由于 Stream<T> 是一个泛型接口，并且无法将primitive用作泛型的类型参数，因此创建了三个新的特殊接口：IntStream, LongStream, DoubleStream。

使用新接口可以减少不必要的自动装箱，从而提高生产力：

IntStream intStream = IntStream.range(1, 3);
LongStream longStream = LongStream.rangeClosed(1, 3);

range(int startInclusive, int endExclusive) 方法创建从第一个参数到第二个参数的有序流。它以等于 1 的步长递增后续元素的值。结果不包括最后一个参数，它只是序列的上限。

rangeClosed(int startInclusive, int endInclusive)方法执行相同的操作，只有一处不同，即包含第二个元素。我们可以使用这两种方法来生成三种类型的primitive流中的任何一种。

从 Java 8 开始，Random 类提供了多种用于生成primitive流的方法。例如，以下代码创建一个 DoubleStream，它具有三个元素：

Random random = new Random();
DoubleStream doubleStream = random.doubles(3);

2.8. 字符串流

我们还可以借助 String 类的 chars() 方法，使用 String 作为创建流的源。由于 JDK 中没有 CharStream 接口，因此我们使用 IntStream 来表示字符流。

IntStream streamOfChars = "abc".chars();

以下示例根据指定的RegEx将字符串分解为子字符串：

Stream<String> streamOfString =
  Pattern.compile(", ").splitAsStream("a, b, c");

2.9.文件流

此外，Java NIO 类 Files 允许我们通过 lines() 方法生成文本文件的Stream<String>。文本的每一行都成为流的一个元素：

Path path = Paths.get("C:\\file.txt");
Stream<String> streamOfStrings = Files.lines(path);
Stream<String> streamWithCharset = 
  Files.lines(path, Charset.forName("UTF-8"));

可以将 Charset 指定为 lines() 方法的参数。

3. 引用流

只要只调用中间操作，我们就可以实例化一个流，并对其进行可访问的引用。执行终端操作导致流不可访问。

为了证明这一点，我们会暂时忘记最佳实践是链接操作顺序。除了不必要的冗长之外，从技术上讲，以下代码是有效的：

Stream<String> stream = 
  Stream.of("a", "b", "c").filter(element -> element.contains("b"));
Optional<String> anyElement = stream.findAny();

但是，在调用终端操作后尝试重用相同的引用将触发 IllegalStateException:

Optional<String> firstElement = stream.findFirst();

由于IllegalStateException是RuntimeException，编译器不会发出有关问题的信号。因此，记住Java 8流不能重用非常重要。

这种行为是合乎逻辑的。我们设计流是为了以函数式的方式将有限的操作序列应用于元素源，而不是存储元素。

因此，为了使之前的代码正常工作，需要进行一些更改：

List<String> elements =
  Stream.of("a", "b", "c").filter(element -> element.contains("b"))
    .collect(Collectors.toList());
Optional<String> anyElement = elements.stream().findAny();
Optional<String> firstElement = elements.stream().findFirst();

4. 流管道

要对数据源的元素执行一系列操作并聚合其结果，我们需要三个部分：源、中间操作和终端操作。

中间操作返回一个新的修改流。例如，要创建一个不含少量元素的现有流的新流，应使用 skip() 方法：

Stream<String> onceModifiedStream =
  Stream.of("abcd", "bbcd", "cbcd").skip(1);

如果我们需要不止一次修改，我们可以链式执行中间操作。假设我们还需要将当前 Stream<String> 中的每个元素替换为包含前几个字符的子字符串。我们可以通过串联skip()和map()方法来实现这一目的：

Stream<String> twiceModifiedStream =
  stream.skip(1).map(element -> element.substring(0, 3));

正如我们所看到的，map() 方法采用 lambda 表达式作为参数。如果我们想了解有关 lambda 的更多信息，可以查看我们的教程Lambda 表达式和函数式接口：提示和最佳实践。

流本身是没有价值的；用户对终端操作的结果感兴趣，该结果可以是某种类型的值或应用于流的每个元素的操作。 每个流只能使用一个终端操作。

使用流的正确且最方便的方法是通过 stream 管道，它是流源、中间操作和终端操作的链：

List<String> list = Arrays.asList("abc1", "abc2", "abc3");
long size = list.stream().skip(1)
  .map(element -> element.substring(0, 3)).sorted().count();

5. 惰性调用

中间操作是惰性的。这意味着只有在终端操作执行需要时才会调用它们。

例如，让我们调用方法 wasCalled()，每次调用时都会增加内部计数器：

private long counter;
 
private void wasCalled() {
    counter++;
}

现在让我们从操作 filter() 中调用方法 wasCalled()：

List<String> list = Arrays.asList(“abc1”, “abc2”, “abc3”);
counter = 0;
Stream<String> stream = list.stream().filter(element -> {
    wasCalled();
    return element.contains("2");
});

由于我们有三个元素的源，我们可以假设 filter() 方法将被调用三次，并且 counter 变量的值为 3。但是，运行这段代码根本不会改变计数器，它仍然为零，因此filter()方法甚至没有被调用一次。原因是缺少终端操作。

让我们稍微重写一下这段代码，添加一个 map() 操作和一个终端操作 findFirst()。我们还将添加借助日志跟踪方法调用顺序的功能：

Optional<String> stream = list.stream().filter(element -> {
    log.info("filter() was called");
    return element.contains("2");
}).map(element -> {
    log.info("map() was called");
    return element.toUpperCase();
}).findFirst();

生成的日志显示我们调用了 filter() 方法两次，调用 map() 方法一次。这是因为管道是垂直执行的。在我们的示例中，流的第一个元素不满足过滤器的谓词。然后我们为第二个元素调用了filter()方法，该方法通过了过滤器。在没有为第三个元素调用 filter() 的情况下，我们通过管道向下到达 map() 方法。

findFirst() 操作仅满足一个元素。因此，在这个特定的示例中，惰性调用使我们能够避免两种方法调用，一种用于 filter()，另一种用于 map()。

六、执行顺序

从性能的角度来看，正确的顺序是流管道中链接操作最重要的方面之一：

long size = list.stream().map(element -> {
    wasCalled();
    return element.substring(0, 3);
}).skip(2).count();

执行此代码将使计数器的值增加 3。这意味着我们调用了流的map()方法3次，但是size的值为1。因此，生成的流只有一个元素，而我们在三次中的两次都无缘无故地执行了昂贵的 map() 操作。

如果我们更改 skip() 和 map() 方法的顺序，计数器将仅增加一。因此我们只会调用map()方法一次：

long size = list.stream().skip(2).map(element -> {
    wasCalled();
    return element.substring(0, 3);
}).count();

这给我们带来了以下规则：减少流大小的中间操作应放置在应用于每个元素的操作之前。因此我们需要保留诸如 skip() 之类的方法、filter()、 和 distinct() 位于流管道的顶部。

7. 减少流量

API 具有许多将流聚合为类型或primitive的终端操作：count()、max()、min() 和 sum()。 但是，这些操作是按照预定义的实现进行的。那么如果开发人员需要自定义 Stream 的还原机制怎么办？有两种方法可以让我们做到这一点，reduce() 和collect() 方法。

7.1. reduce() 方法

此方法有三种变体，其签名和返回类型有所不同。它们可以具有以下参数：

身份 - 累加器的初始值，如果数据流为空，没有任何东西需要累加，则为默认值

累加器（accumulator） - 指定元素聚合逻辑的函数。由于累加器每减少一步就会创建一个新值，因此新值的数量等于数据流的大小，只有最后一个值才有用。这对性能来说不是很好。

combiner - 一个汇总累加器结果的函数。我们只在并行模式下调用组合器，以减少来自不同线程的累加器结果。

现在让我们看看这三种方法的实际效果：

OptionalInt reduced =
  IntStream.range(1, 4).reduce((a, b) -> a + b);

reduced = 6 (1 + 2 + 3)

int reducedTwoParams =
  IntStream.range(1, 4).reduce(10, (a, b) -> a + b);

reducedTwoParams = 16 (10 + 1 + 2 + 3)

int reducedParams = Stream.of(1, 2, 3)
  .reduce(10, (a, b) -> a + b, (a, b) -> {
     log.info("combiner was called");
     return a + b;
  });

结果将与前面的示例 (16) 相同，并且不会登录，这意味着组合器没有被调用。为了使组合器工作，流应该是并行的：

int reducedParallel = Arrays.asList(1, 2, 3).parallelStream()
    .reduce(10, (a, b) -> a + b, (a, b) -> {
       log.info("combiner was called");
       return a + b;
    });

这里的结果是不同的（36），并且组合器被调用了两次。这里的 reduction 是通过以下算法进行的：累加器通过将流的每个元素添加到identity来运行三次。这些行动是并行进行的。结果是 (10 + 1 = 11; 10 + 2 = 12; 10 + 3 = 13;)。现在combiner可以合并这三个结果。为此需要两次迭代（12 + 13 = 25；25 + 11 = 36）。

7.2. collect() 方法

流的reduction 也可以通过另一个终端操作来执行，即collect()方法。它接受Collector 类型的参数，它指定reduction机制。大多数常见操作都有预定义的收集器。可以借助Collectors类型来访问它们。

在本节中，我们将使用以下列表作为所有流的源：

List<Product> productList = Arrays.asList(new Product(23, "potatoes"),
  new Product(14, "orange"), new Product(13, "lemon"),
  new Product(23, "bread"), new Product(13, "sugar"));

将流转换为集合（集合、列表或集合）：

List<String> collectorCollection = 
  productList.stream().map(Product::getName).collect(Collectors.toList());

缩减为字符串：

String listToString = productList.stream().map(Product::getName)
  .collect(Collectors.joining(", ", "[", "]"));

joiner() 方法可以具有一到三个参数（分隔符、前缀、后缀）。使用 joiner() 最方便的地方在于，开发人员无需检查数据流是否到达终点，也无需应用后缀和分隔符。 Collector会处理这个问题。

处理流中所有数字元素的平均值：

double averagePrice = productList.stream()
  .collect(Collectors.averagingInt(Product::getPrice));

处理流中所有数字元素的总和：

int summingPrice = productList.stream()
  .collect(Collectors.summingInt(Product::getPrice));

averagingXX()、summingXX() 和 summarizingXX() 方法可以与基元（int、long、double）及其包装类（Integer、Long、Double）一起使用。这些方法的一项更强大的功能是提供映射。因此，开发人员不需要在 collect() 方法之前使用额外的 map() 操作。

收集有关流元素的统计信息：

IntSummaryStatistics statistics = productList.stream()
  .collect(Collectors.summarizingInt(Product::getPrice));

通过使用 IntSummaryStatistics 类型的结果实例，开发人员可以通过应用 toString() 方法创建统计报告。结果将是一个字符串，与此字符串“IntSummaryStatistics{count=5, sum=86, min=13, average=17,200000, max=23}.”相同。

通过应用 getCount()、getSum()、getMin() 方法，也可以轻松从此对象中提取 count、sum、min、 和 average 的单独值、 getAverage()、和 getMax()。 所有这些值都可以从单个管道中提取。

根据指定的函数对流的元素进行分组：

Map<Integer, List<Product>> collectorMapOfLists = productList.stream()
  .collect(Collectors.groupingBy(Product::getPrice));

在上面的示例中，流被简化为Map，它按价格对所有产品进行分组。

根据某些谓词将流的元素分组：

Map<Boolean, List<Product>> mapPartioned = productList.stream()
  .collect(Collectors.partitioningBy(element -> element.getPrice() > 15));

推动收集器执行额外的转换：

Set<Product> unmodifiableSet = productList.stream()
  .collect(Collectors.collectingAndThen(Collectors.toSet(),
  Collections::unmodifiableSet));

在这种特殊情况下，收集器已将流转换为Set，然后从中创建不可更改的Set。

自定义收集器：

如果由于某种原因应创建自定义收集器，最简单且最简单的方法是使用 Collector 类型的方法 of()。

Collector<Product, ?, LinkedList<Product>> toLinkedList =
  Collector.of(LinkedList::new, LinkedList::add, 
    (first, second) -> { 
       first.addAll(second); 
       return first; 
    });

LinkedList<Product> linkedListOfPersons =
  productList.stream().collect(toLinkedList);

在这个例子中，Collector 的一个实例被简化为 LinkedList<Persone>。

8. 并行流

在 Java 8 之前，并行化很复杂。 ExecutorService 和ForkJoin 的出现简化了开发人员的工作，但如何创建特定的执行器、如何运行执行器等问题仍然值得记忆。 Java 8 引入了一种以函数式风格实现并行性的方法。

API 允许我们创建并行流，以并行模式执行操作。当流的源是Collection或array时，可以借助parallelStream()来实现方法：

Stream<Product> streamOfCollection = productList.parallelStream();
boolean isParallel = streamOfCollection.isParallel();
boolean bigPrice = streamOfCollection
  .map(product -> product.getPrice() * 12)
  .anyMatch(price -> price > 200);

如果流的源不是集合或数组，则parallel()方法应该使用：

IntStream intStreamParallel = IntStream.range(1, 150).parallel();
boolean isParallel = intStreamParallel.isParallel();

在底层，Stream API 自动使用 ForkJoin 框架来并行执行操作。默认情况下，将使用公共线程池，并且无法（至少目前）为其分配一些自定义线程池。 可以通过使用一组自定义并行收集器来克服这个问题。

以并行模式使用流时，避免阻塞操作。当任务需要相似的执行时间时，最好使用并行模式。如果一项任务比另一项任务持续的时间长得多，则可能会减慢整个应用程序的工作流程。

并行模式下的流可以使用 sequential() 方法转换回顺序模式：

IntStream intStreamSequential = intStreamParallel.sequential();
boolean isParallel = intStreamSequential.isParallel();

9. 结论

Stream API 是一组功能强大但易于理解的工具，用于处理元素序列。如果使用得当，它可以让我们减少大量的样板代码，创建更具可读性的程序，并提高应用程序的生产力。

在本文中显示的大多数代码示例中，我们未使用流（我们没有应用 close() 方法或终端操作）。在真实的应用程序中，不要留下未使用的实例化流，因为这会导致内存泄漏。

本文随附的完整代码示例可在 GitHub 上获取。