摘要：處理數據流時，可以避免在不必要時檢查所有數據。它對每個輸入元素執行一個映射函數并返回一個包含結果值統計信息的特殊類。例如等操作可能在并行數據流中產生不同的結果。

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()有兩個參數：一個初始值，稱為種子值，和一個使用前一個值來生成下一個值的函數。該方法是有狀態的，因此不適合并行運行。

想要了解更多開發技術，面試教程以及互聯網公司內推，歡迎關注我的微信公眾號！將會不定期的發放福利哦~