猫头鹰的深夜翻译：Java Streams

摘要：处理数据流时，可以避免在不必要时检查所有数据。它对每个输入元素执行一个映射函数并返回一个包含结果值统计信息的特殊类。例如等操作可能在并行数据流中产生不同的结果。

Stream是Java 8中的一个重大新功能。这个深入的教程是流支持的许多功能的介绍，并着重于简单实用的示例。

了解这个之前，你需要有对Java 8有基础的实践性的知识(lambda表达式，方法引用)

首先，不应该将Java 8 Streams与Java I/O流混淆（例如：FileInputStream等）;它们彼此之间关联不大。

简而言之，流是对数据源的包装，使我们能够使用该数据源并对其快速便捷的批量处理。

流不存储数据，从这个意义上说，它不是一个数据结构。它也从不修改底层数据源。

java.util.stream提供的新功能支持对元素的流进行函数式编程风格的操作，比如在集合上进行map-reduce转换。

现在让我们在讨论术语和核心概念之前，深入一些简单的流创建和使用示例。

我们首先从现有数组中获取流：

private static Employee[] arrayOfEmps = {
    new Employee(1, "Jeff Bezos", 100000.0), 
    new Employee(2, "Bill Gates", 200000.0), 
    new Employee(3, "Mark Zuckerberg", 300000.0)
};
Stream.of(arrayOfEmps);

我们也能从现有的列表中获取流：

private static List empList = Arrays.asList(arrayOfEmps);
empList.stream();

请注意，Java 8在Collection接口添加了一个新的stream()方法。

我们也可以在独立对象上使用Stream.of()创建流:

Stream.of(arrayOfEmps[0], arrayOfEmps[1], arrayOfEmps[2]);

或者直接使用Stream.builder():

Stream.Builder empStreamBuilder = Stream.builder();
empStreamBuilder.accept(arrayOfEmps[0]);
empStreamBuilder.accept(arrayOfEmps[1]);
empStreamBuilder.accept(arrayOfEmps[2]);
Stream empStream = empStreamBuilder.build();

还有其他方法可以获得流，其中的一些方法我们将在下面的部分中看到。

现在让我们看看我们可以在语言中使用新流支持的帮助下执行的一些常见用法和操作。

forEach()是最简单也是最常用的操作。它遍历流元素，并在每个元素上调用提供的函数。

这个方法非常常见，它直接在Iterable，Map等中引入了：

@Test
public void whenIncrementSalaryForEachEmployee_thenApplyNewSalary() {    
    empList.stream().forEach(e -> e.salaryIncrement(10.0));
    assertThat(empList, contains(
      hasProperty("salary", equalTo(110000.0)),
      hasProperty("salary", equalTo(220000.0)),
      hasProperty("salary", equalTo(330000.0))
    ));
}

它将有效地调用empList中每个元素的salaryIncrement()方法。

forEach()是一个终结操作。在执行该操作后，流管道将被视为已经被使用，将无法再被使用。我们会在下一节继续讨论终结操作。

map()在对原始流执行完函数后会创建一个新的流。新的流将会是另一种类型。

以下示例将整数流转换为Employee流：

@Test
public void whenMapIdToEmployees_thenGetEmployeeStream() {
    Integer[] empIds = { 1, 2, 3 };
    List employees = Stream.of(empIds)
      .map(employeeRepository::findById)
      .collect(Collectors.toList());
    assertEquals(employees.size(), empIds.length);
}

这里我们先从一个数组中获得员工Id流。每个Id被传入employeeRepository:findById()方法并返回对应Employee对象，从而高效的生成一个员工流。

我们可以看到collect()方法是如何在前一个例子中工作的。当我们完成所有处理，就可以用这种方法从流中获取内容：

@Test
public void whenCollectStreamToList_thenGetList() {
    List employees = empList.stream().collect(Collectors.toList());
    assertEquals(empList, employees);
}

collect()对流中的数据元素执行可变折叠操作(将元素重新打包到一些数据结构并进行一些额外的逻辑处理，比如将它们连接起来)。

此操作的策略通过Collections接口的实现来提供。在上面的例子中，我们使用toList收集器将流中的元素收集到List实例中。

现在让我们看一下filter()方法。这会产生一个新的流，其中包含通过给定测试（由Predicate指定）的原始流的元素。

@Test
public void whenFilterEmployees_thenGetFilteredStream() {
    Integer[] empIds = { 1, 2, 3, 4 };
    List employees = Stream.of(empIds)
      .map(employeeRepository::findById)
      .filter(e -> e != null)
      .filter(e -> e.getSalary() > 200000)
      .collect(Collectors.toList());
    assertEquals(Arrays.asList(arrayOfEmps[2]), employees);
}

在上面的例子中，我们先筛选掉值为null的不合法员工号，然后再使用了一个过滤器筛选出工资超过一定阈值的员工。

@Test
public void whenFindFirst_thenGetFirstEmployeeInStream() {
    Integer[] empIds = { 1, 2, 3, 4 };
    Employee employee = Stream.of(empIds)
      .map(employeeRepository::findById)
      .filter(e -> e != null)
      .filter(e -> e.getSalary() > 100000)
      .findFirst()
      .orElse(null);
    assertEquals(employee.getSalary(), new Double(200000));
}

这里会返回薪水大于10000的第一个员工，如果没有薪水大于10000的员工，则返回null。

我们已经看到了如何使用collect()从数据流中获取数据。如果我们需要从流中获取数组，我们可以简单地使用toArray()：

@Test
public void whenStreamToArray_thenGetArray() {
    Employee[] employees = empList.stream().toArray(Employee[]::new);
    assertThat(empList.toArray(), equalTo(employees));
}             

Employee[]::new会新建一个空的Employee数组 - 它会用流中的元素填充。

流可以容纳复杂的数据结构，如Stream>。在这样的场景下，我们可以使用flatMap()来扁平化数据结构，简化后序的操作。

@Test
public void whenFlatMapEmployeeNames_thenGetNameStream() {
    List> namesNested = Arrays.asList( 
      Arrays.asList("Jeff", "Bezos"), 
      Arrays.asList("Bill", "Gates"), 
      Arrays.asList("Mark", "Zuckerberg"));
    List namesFlatStream = namesNested.stream()
      .flatMap(Collection::stream)
      .collect(Collectors.toList());
    assertEquals(namesFlatStream.size(), namesNested.size() * 2);
}

我们在前面看到了forEach()的使用，它是一个终结操作。但是，有时我们需要在执行任何终结操作之前对流的每个元素执行多个操作。

peek()方法在这种场景下很实用。简单来说，它会在流的每个元素上执行特定的操作，并返回一个新的流。peek()是一个中间操作。

@Test
public void whenIncrementSalaryUsingPeek_thenApplyNewSalary() {
    Employee[] arrayOfEmps = {
        new Employee(1, "Jeff Bezos", 100000.0), 
        new Employee(2, "Bill Gates", 200000.0), 
        new Employee(3, "Mark Zuckerberg", 300000.0)
    };
    List empList = Arrays.asList(arrayOfEmps);
    empList.stream()
      .peek(e -> e.salaryIncrement(10.0))
      .peek(System.out::println)
      .collect(Collectors.toList());
    assertThat(empList, contains(
      hasProperty("salary", equalTo(110000.0)),
      hasProperty("salary", equalTo(220000.0)),
      hasProperty("salary", equalTo(330000.0))
    ));
}

我们之前讨论时提出，流操作可以分为中间操作和终结操作。

中间操作如filter()会返回一个新的流，并且可以在该流之上进行后续操作。终结操作如forEach()，将流标记为已经使用，在这之后该流就不可以被使用了。

一个流管道由一个流源组成，然后是零个或多个中间操作，以及一个终端操作。

@Test
public void whenStreamCount_thenGetElementCount() {
    Long empCount = empList.stream()
      .filter(e -> e.getSalary() > 200000)
      .count();
    assertEquals(empCount, new Long(1));
}

一些操作被定义为短路操作，短路操作允许在无尽流上的计算可以在有限时间内完成:

@Test
public void whenLimitInfiniteStream_thenGetFiniteElements() {
    Stream infiniteStream = Stream.iterate(2, i -> i * 2);
    List collect = infiniteStream
      .skip(3)
      .limit(5)
      .collect(Collectors.toList());
    assertEquals(collect, Arrays.asList(16, 32, 64, 128, 256));
}

我们会在后面继续讨论无尽流。

流的最重要的特征之一是它们允许通过惰性计算进行显著的优化。只有在启动终结操作的时候才会执行流上的计算。所有的中间操作都是惰性执行的，所以除非在需要得出结果的时候，否则它们不会执行。

比如，我们之前看到的findFirst()例子。这里执行了多少次map()操作？4次，因为输入数组包含4个元素。

@Test
public void whenFindFirst_thenGetFirstEmployeeInStream() {
    Integer[] empIds = { 1, 2, 3, 4 };
    Employee employee = Stream.of(empIds)
      .map(employeeRepository::findById)
      .filter(e -> e != null)
      .filter(e -> e.getSalary() > 100000)
      .findFirst()
      .orElse(null);
    assertEquals(employee.getSalary(), new Double(200000));
}

Stream执行了一个map操作和两个filter操作。

它首先在id 1上执行所有操作。由于id 1的工资不大于100000，处理转移到下一个元素。

Id 2满足两个过滤器谓词，因此流将执行终结操作findFirst()并返回结果。

在Id 3 和Id 4上不会执行任何操作。

处理数据流时，可以避免在不必要时检查所有数据。当输入流是无限的并且非常大时，这种行为变得更加重要。

让我们从sorted()方法开始。它会根据我们传入的比较器对流元素进行排序。

例如，我们可以根据名字对员工进行排序：

@Test
public void whenSortStream_thenGetSortedStream() {
    List employees = empList.stream()
      .sorted((e1, e2) -> e1.getName().compareTo(e2.getName()))
      .collect(Collectors.toList());
    assertEquals(employees.get(0).getName(), "Bill Gates");
    assertEquals(employees.get(1).getName(), "Jeff Bezos");
    assertEquals(employees.get(2).getName(), "Mark Zuckerberg");
}

需要注意在sorted()方法中不会进行短路操作。

@Test
public void whenFindMin_thenGetMinElementFromStream() {
    Employee firstEmp = empList.stream()
      .min((e1, e2) -> e1.getId() - e2.getId())
      .orElseThrow(NoSuchElementException::new);
    assertEquals(firstEmp.getId(), new Integer(1));
}

我们还可以通过使用Comparator.comparing()方法免去定义比较逻辑。

@Test
public void whenFindMax_thenGetMaxElementFromStream() {
    Employee maxSalEmp = empList.stream()
      .max(Comparator.comparing(Employee::getSalary))
      .orElseThrow(NoSuchElementException::new);
    assertEquals(maxSalEmp.getSalary(), new Double(300000.0));
}

distinct()不接受任何参数并返回流中的不同元素，从而消除重复。它使用元素的equals()方法来决定两个元素是否相等：

@Test
public void whenApplyDistinct_thenRemoveDuplicatesFromStream() {
    List intList = Arrays.asList(2, 5, 3, 2, 4, 3);
    List distinctIntList = intList.stream().distinct().collect(Collectors.toList());
    assertEquals(distinctIntList, Arrays.asList(2, 5, 3, 4));
}

这些操作会接收一个Predicate并返回一个boolean值。一旦确定了答案，就执行短路操作并停止处理：

@Test
public void whenApplyMatch_thenReturnBoolean() {
    List intList = Arrays.asList(2, 4, 5, 6, 8);
    boolean allEven = intList.stream().allMatch(i -> i % 2 == 0);
    boolean oneEven = intList.stream().anyMatch(i -> i % 2 == 0);
    boolean noneMultipleOfThree = intList.stream().noneMatch(i -> i % 3 == 0);
    assertEquals(allEven, false);
    assertEquals(oneEven, true);
    assertEquals(noneMultipleOfThree, false);
}

allMatch()会检查流中所有元素的谓词是否为真。在它遇到5时无法被2整除，它会立即返回false。

anyMatch()会检查流中任何一个元素的谓词是否为真。这里，再次施加短路操作并且在第一个元素之后立即返回true。

noneMatch()检查是否没有与谓词匹配的元素。在这里，只要遇到可被3整除的6就返回false。

目前为止，我们讨论的都是对象引用流。但是还有IntStream, LongStream, 和 DoubleStream分别对应Int，Long和Double基础数据类型的流。当需要处理大量的数字类型值时，使用它们会非常方便。

创建一个IntStream最常用的方法是在一个现有流上调用mapToInt()方法。

@Test
public void whenFindMaxOnIntStream_thenGetMaxInteger() {
    Integer latestEmpId = empList.stream()
      .mapToInt(Employee::getId)
      .max()
      .orElseThrow(NoSuchElementException::new);
    assertEquals(latestEmpId, new Integer(3));
}

我们先生成了一个empList的流然后再在其上通过在mapToInt中调用Employee::getId方法来获得一个IntStream。最后我们调用max()获得最大值。

我们还可以使用IntStream.of()生成IntStream

IntStream.of(1, 2, 3);

或者是IntStream.range():

IntStream.range(10, 20)

它会生成一个包含10-19之间所有整数的IntStream。
这里有一个重要的区别需要注意一下：

Stream.of(1, 2, 3)

该方法生成的是一个Stream对象而不是IntStream。
类似的，使用map()而不是mapToInt()将会生成Stream而不是IntStream。

empList.stream().map(Employee::getId);

特定类型的流相比于标准的流提供了额外的操作。比如sum(), average(), range()等。

@Test
public void whenApplySumOnIntStream_thenGetSum() {
    Double avgSal = empList.stream()
      .mapToDouble(Employee::getSalary)
      .average()
      .orElseThrow(NoSuchElementException::new);
    assertEquals(avgSal, new Double(200000));
}

Reduction操作（也称为fold）获得一系列输入元素，并通过重复执行组合操作将它们组合为单个汇总结果。我们已经看到过几个Reduction操作如findFirst(), min()和max()。

让我们看一看通俗意义上的reduce()的使用。

T reduce(T identity, BinaryOperator accumulator)

identity代表起始值而accumulator代表一个二元操作符。

@Test
public void whenApplyReduceOnStream_thenGetValue() {
    Double sumSal = empList.stream()
      .map(Employee::getSalary)
      .reduce(0.0, Double::sum);
    assertEquals(sumSal, new Double(600000));
}

这里我们将起始值设置为0.0并且对流上的元素重复的执行Double::sum()。通过在Stream中使用reduce我们有效的实现了DoubleStream的sum方法。

我们已经看过如何使用Collectors.toList()从流中获取list。让我们再看几个从流中获取数据的方法。

@Test
public void whenCollectByJoining_thenGetJoinedString() {
    String empNames = empList.stream()
      .map(Employee::getName)
      .collect(Collectors.joining(", "))
      .toString();
    assertEquals(empNames, "Jeff Bezos, Bill Gates, Mark Zuckerberg");
}

我们还可以使用toSet方法从流中获取Set：

@Test
public void whenCollectBySet_thenGetSet() {
    Set empNames = empList.stream()
            .map(Employee::getName)
            .collect(Collectors.toSet());
    assertEquals(empNames.size(), 3);
}

@Test
public void whenToVectorCollection_thenGetVector() {
    Vector empNames = empList.stream()
            .map(Employee::getName)
            .collect(Collectors.toCollection(Vector::new));
    assertEquals(empNames.size(), 3);
}

这里内部创建了一个新的空集合，并对流中的每个元素调用了add()方法。

summarizingDouble是另一个有趣的收集器。它对每个输入元素执行一个double-producing映射函数并返回一个包含结果值统计信息的特殊类。

@Test
public void whenApplySummarizing_thenGetBasicStats() {
    DoubleSummaryStatistics stats = empList.stream()
      .collect(Collectors.summarizingDouble(Employee::getSalary));
    assertEquals(stats.getCount(), 3);
    assertEquals(stats.getSum(), 600000.0, 0);
    assertEquals(stats.getMin(), 100000.0, 0);
    assertEquals(stats.getMax(), 300000.0, 0);
    assertEquals(stats.getAverage(), 200000.0, 0);
}

可以看到我们是如何分析每位员工的工资并获取有关该数据的统计信息 - 如最小值，最大值，平均值等。

summaryStatistics()可以在使用特定流的时候用来生成类似的结果：

@Test
public void whenApplySummaryStatistics_thenGetBasicStats() {
    DoubleSummaryStatistics stats = empList.stream()
      .mapToDouble(Employee::getSalary)
      .summaryStatistics();
    assertEquals(stats.getCount(), 3);
    assertEquals(stats.getSum(), 600000.0, 0);
    assertEquals(stats.getMin(), 100000.0, 0);
    assertEquals(stats.getMax(), 300000.0, 0);
    assertEquals(stats.getAverage(), 200000.0, 0);
}

我们可以根据元素是否满足某个条例将一个流分解为两个。

让我们将一个数值数组分成奇数数组和偶数数组：

@Test
public void whenStreamPartition_thenGetMap() {
    List intList = Arrays.asList(2, 4, 5, 6, 8);
    Map> isEven = intList.stream().collect(
      Collectors.partitioningBy(i -> i % 2 == 0));
    assertEquals(isEven.get(true).size(), 4);
    assertEquals(isEven.get(false).size(), 1);
}

在这里，流被分解并存入Map中，并使用true和false键代表偶数数组和奇数数组。

groupingBy()提供高级分解。它将我们的流分解为两个或多个子流。

它接收一个分类方法作为参数。这个分类方法会作用于流中的每一个元素。

分类方法返回的值会作为Map的键。

@Test
public void whenStreamGroupingBy_thenGetMap() {
    Map> groupByAlphabet = empList.stream().collect(
      Collectors.groupingBy(e -> new Character(e.getName().charAt(0))));
    assertEquals(groupByAlphabet.get("B").get(0).getName(), "Bill Gates");
    assertEquals(groupByAlphabet.get("J").get(0).getName(), "Jeff Bezos");
    assertEquals(groupByAlphabet.get("M").get(0).getName(), "Mark Zuckerberg");
}

在上面这个简单的例子中，我们根据员工的首字母进行分组。groupingBy()使用Map对流中的数据进行分组。但是，有时候我们可能需要将元素分组为另一种类型。我们可以使用mapping()，它实际上可以使收集器适应不同的类型。

@Test
public void whenStreamMapping_thenGetMap() {
    Map> idGroupedByAlphabet = empList.stream().collect(
      Collectors.groupingBy(e -> new Character(e.getName().charAt(0)),
        Collectors.mapping(Employee::getId, Collectors.toList())));
    assertEquals(idGroupedByAlphabet.get("B").get(0), new Integer(2));
    assertEquals(idGroupedByAlphabet.get("J").get(0), new Integer(1));
    assertEquals(idGroupedByAlphabet.get("M").get(0), new Integer(3));
}

这里mapping()使用getId()映射函数将流元素Employee映射到员工id - 这是一个Integer。这些ID仍然根据员工名字的首字母进行分组。
reducing()类似于reduce():

@Test
public void whenStreamReducing_thenGetValue() {
    Double percentage = 10.0;
    Double salIncrOverhead = empList.stream().collect(Collectors.reducing(
        0.0, e -> e.getSalary() * percentage / 100, (s1, s2) -> s1 + s2));
    assertEquals(salIncrOverhead, 60000.0, 0);
}

reducing + groupingBy

@Test
public void whenStreamGroupingAndReducing_thenGetMap() {
    Comparator byNameLength = Comparator.comparing(Employee::getName);
    Map> longestNameByAlphabet = empList.stream().collect(
      Collectors.groupingBy(e -> new Character(e.getName().charAt(0)),
        Collectors.reducing(BinaryOperator.maxBy(byNameLength))));
    assertEquals(longestNameByAlphabet.get("B").get().getName(), "Bill Gates");
    assertEquals(longestNameByAlphabet.get("J").get().getName(), "Jeff Bezos");
    assertEquals(longestNameByAlphabet.get("M").get().getName(), "Mark Zuckerberg");
}

我们首先根据员工的首字母将其分组，然后在各个组里，我们找到名字最长的员工。

@Test
public void whenParallelStream_thenPerformOperationsInParallel() {
    Employee[] arrayOfEmps = {
      new Employee(1, "Jeff Bezos", 100000.0), 
      new Employee(2, "Bill Gates", 200000.0), 
      new Employee(3, "Mark Zuckerberg", 300000.0)
    };
    List empList = Arrays.asList(arrayOfEmps);
    empList.stream().parallel().forEach(e -> e.salaryIncrement(10.0));
    assertThat(empList, contains(
      hasProperty("salary", equalTo(110000.0)),
      hasProperty("salary", equalTo(220000.0)),
      hasProperty("salary", equalTo(330000.0))
    ));
}

因为这里涉及到多线程，所以我们需要注意一下几点：

确保代码是线程安全的。特别要注意并行操作可能会的修改的共享数据。

如果执行操作的顺序或输出流中返回的顺序很重要，我们不应该使用并行流。例如findFirst()等操作可能在并行数据流中产生不同的结果。

确保并行执行是值得的。

@Test
public void whenGenerateStream_thenGetInfiniteStream() {
    Stream.generate(Math::random)
      .limit(5)
      .forEach(System.out::println);
}

我们给generate()方法提供了Supplier，当需要新元素时就会调用这个方法。
我们需要提供一个终止进程的条件。通常使用的一种方法是limit()。在上面的例子中，我们将元素的数量限制为5，并在它们生成时候打印它们。

@Test
public void whenIterateStream_thenGetInfiniteStream() {
    Stream evenNumStream = Stream.iterate(2, i -> i * 2);
    List collect = evenNumStream
      .limit(5)
      .collect(Collectors.toList());
    assertEquals(collect, Arrays.asList(2, 4, 8, 16, 32));
}

iterate()有两个参数：一个初始值，称为种子值，和一个使用前一个值来生成下一个值的函数。该方法是有状态的，因此不适合并行运行。

想要了解更多开发技术，面试教程以及互联网公司内推，欢迎关注我的微信公众号！将会不定期的发放福利哦~