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

Stream是Java 8中的一個重大新功能。這個深入的教程是流支持的許多功能的介紹，并著重于簡單實用的示例。

了解這個之前，你需要有對Java 8有基礎(chǔ)的實踐性的知識(lambda表達式，方法引用)

首先，不應該將Java 8 Streams與Java I/O流混淆（例如：FileInputStream等）;它們彼此之間關(guān)聯(lián)不大。

簡而言之，流是對數(shù)據(jù)源的包裝，使我們能夠使用該數(shù)據(jù)源并對其快速便捷的批量處理。

流不存儲數(shù)據(jù)，從這個意義上說，它不是一個數(shù)據(jù)結(jié)構(gòu)。它也從不修改底層數(shù)據(jù)源。

java.util.stream提供的新功能支持對元素的流進行函數(shù)式編程風格的操作，比如在集合上進行map-reduce轉(zhuǎn)換。

現(xiàn)在讓我們在討論術(shù)語和核心概念之前，深入一些簡單的流創(chuàng)建和使用示例。

我們首先從現(xiàn)有數(shù)組中獲取流：

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);

我們也能從現(xiàn)有的列表中獲取流：

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

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

我們也可以在獨立對象上使用Stream.of()創(chuàng)建流:

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();

還有其他方法可以獲得流，其中的一些方法我們將在下面的部分中看到。

現(xiàn)在讓我們看看我們可以在語言中使用新流支持的幫助下執(zhí)行的一些常見用法和操作。

forEach()是最簡單也是最常用的操作。它遍歷流元素，并在每個元素上調(diào)用提供的函數(shù)。

這個方法非常常見，它直接在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))
    ));
}

它將有效地調(diào)用empList中每個元素的salaryIncrement()方法。

forEach()是一個終結(jié)操作。在執(zhí)行該操作后，流管道將被視為已經(jīng)被使用，將無法再被使用。我們會在下一節(jié)繼續(xù)討論終結(jié)操作。

map()在對原始流執(zhí)行完函數(shù)后會創(chuàng)建一個新的流。新的流將會是另一種類型。

以下示例將整數(shù)流轉(zhuǎn)換為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);
}

這里我們先從一個數(shù)組中獲得員工Id流。每個Id被傳入employeeRepository:findById()方法并返回對應Employee對象，從而高效的生成一個員工流。

我們可以看到collect()方法是如何在前一個例子中工作的。當我們完成所有處理，就可以用這種方法從流中獲取內(nèi)容：

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

collect()對流中的數(shù)據(jù)元素執(zhí)行可變折疊操作(將元素重新打包到一些數(shù)據(jù)結(jié)構(gòu)并進行一些額外的邏輯處理，比如將它們連接起來)。

此操作的策略通過Collections接口的實現(xiàn)來提供。在上面的例子中，我們使用toList收集器將流中的元素收集到List實例中。

現(xiàn)在讓我們看一下filter()方法。這會產(chǎn)生一個新的流，其中包含通過給定測試（由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。

我們已經(jīng)看到了如何使用collect()從數(shù)據(jù)流中獲取數(shù)據(jù)。如果我們需要從流中獲取數(shù)組，我們可以簡單地使用toArray()：

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

Employee[]::new會新建一個空的Employee數(shù)組 - 它會用流中的元素填充。

流可以容納復雜的數(shù)據(jù)結(jié)構(gòu)，如Stream>。在這樣的場景下，我們可以使用flatMap()來扁平化數(shù)據(jù)結(jié)構(gòu)，簡化后序的操作。

@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()的使用，它是一個終結(jié)操作。但是，有時我們需要在執(zhí)行任何終結(jié)操作之前對流的每個元素執(zhí)行多個操作。

peek()方法在這種場景下很實用。簡單來說，它會在流的每個元素上執(zhí)行特定的操作，并返回一個新的流。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))
    ));
}

我們之前討論時提出，流操作可以分為中間操作和終結(jié)操作。

中間操作如filter()會返回一個新的流，并且可以在該流之上進行后續(xù)操作。終結(jié)操作如forEach()，將流標記為已經(jīng)使用，在這之后該流就不可以被使用了。

一個流管道由一個流源組成，然后是零個或多個中間操作，以及一個終端操作。

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

一些操作被定義為短路操作，短路操作允許在無盡流上的計算可以在有限時間內(nèi)完成:

@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));
}

我們會在后面繼續(xù)討論無盡流。

流的最重要的特征之一是它們允許通過惰性計算進行顯著的優(yōu)化。只有在啟動終結(jié)操作的時候才會執(zhí)行流上的計算。所有的中間操作都是惰性執(zhí)行的，所以除非在需要得出結(jié)果的時候，否則它們不會執(zhí)行。

比如，我們之前看到的findFirst()例子。這里執(zhí)行了多少次map()操作？4次，因為輸入數(shù)組包含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執(zhí)行了一個map操作和兩個filter操作。

它首先在id 1上執(zhí)行所有操作。由于id 1的工資不大于100000，處理轉(zhuǎn)移到下一個元素。

Id 2滿足兩個過濾器謂詞，因此流將執(zhí)行終結(jié)操作findFirst()并返回結(jié)果。

在Id 3 和Id 4上不會執(zhí)行任何操作。

處理數(shù)據(jù)流時，可以避免在不必要時檢查所有數(shù)據(jù)。當輸入流是無限的并且非常大時，這種行為變得更加重要。

讓我們從sorted()方法開始。它會根據(jù)我們傳入的比較器對流元素進行排序。

例如，我們可以根據(jù)名字對員工進行排序：

@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()不接受任何參數(shù)并返回流中的不同元素，從而消除重復。它使用元素的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值。一旦確定了答案，就執(zhí)行短路操作并停止處理：

@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基礎(chǔ)數(shù)據(jù)類型的流。當需要處理大量的數(shù)字類型值時，使用它們會非常方便。

創(chuàng)建一個IntStream最常用的方法是在一個現(xiàn)有流上調(diào)用mapToInt()方法。

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

我們先生成了一個empList的流然后再在其上通過在mapToInt中調(diào)用Employee::getId方法來獲得一個IntStream。最后我們調(diào)用max()獲得最大值。

我們還可以使用IntStream.of()生成IntStream

IntStream.of(1, 2, 3);

或者是IntStream.range():

IntStream.range(10, 20)

它會生成一個包含10-19之間所有整數(shù)的IntStream。
這里有一個重要的區(qū)別需要注意一下：

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）獲得一系列輸入元素，并通過重復執(zhí)行組合操作將它們組合為單個匯總結(jié)果。我們已經(jīng)看到過幾個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));
}

這里我們將起始值設(shè)置為0.0并且對流上的元素重復的執(zhí)行Double::sum()。通過在Stream中使用reduce我們有效的實現(xiàn)了DoubleStream的sum方法。

我們已經(jīng)看過如何使用Collectors.toList()從流中獲取list。讓我們再看幾個從流中獲取數(shù)據(jù)的方法。

@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);
}

這里內(nèi)部創(chuàng)建了一個新的空集合，并對流中的每個元素調(diào)用了add()方法。

summarizingDouble是另一個有趣的收集器。它對每個輸入元素執(zhí)行一個double-producing映射函數(shù)并返回一個包含結(jié)果值統(tǒng)計信息的特殊類。

@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);
}

可以看到我們是如何分析每位員工的工資并獲取有關(guān)該數(shù)據(jù)的統(tǒng)計信息 - 如最小值，最大值，平均值等。

summaryStatistics()可以在使用特定流的時候用來生成類似的結(jié)果：

@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);
}

我們可以根據(jù)元素是否滿足某個條例將一個流分解為兩個。

讓我們將一個數(shù)值數(shù)組分成奇數(shù)數(shù)組和偶數(shù)數(shù)組：

@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鍵代表偶數(shù)數(shù)組和奇數(shù)數(shù)組。

groupingBy()提供高級分解。它將我們的流分解為兩個或多個子流。

它接收一個分類方法作為參數(shù)。這個分類方法會作用于流中的每一個元素。

分類方法返回的值會作為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");
}

在上面這個簡單的例子中，我們根據(jù)員工的首字母進行分組。groupingBy()使用Map對流中的數(shù)據(jù)進行分組。但是，有時候我們可能需要將元素分組為另一種類型。我們可以使用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()映射函數(shù)將流元素Employee映射到員工id - 這是一個Integer。這些ID仍然根據(jù)員工名字的首字母進行分組。
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");
}

我們首先根據(jù)員工的首字母將其分組，然后在各個組里，我們找到名字最長的員工。

@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))
    ));
}

因為這里涉及到多線程，所以我們需要注意一下幾點：

確保代碼是線程安全的。特別要注意并行操作可能會的修改的共享數(shù)據(jù)。

如果執(zhí)行操作的順序或輸出流中返回的順序很重要，我們不應該使用并行流。例如findFirst()等操作可能在并行數(shù)據(jù)流中產(chǎn)生不同的結(jié)果。

確保并行執(zhí)行是值得的。

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

我們給generate()方法提供了Supplier，當需要新元素時就會調(diào)用這個方法。
我們需要提供一個終止進程的條件。通常使用的一種方法是limit()。在上面的例子中，我們將元素的數(shù)量限制為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()有兩個參數(shù)：一個初始值，稱為種子值，和一個使用前一個值來生成下一個值的函數(shù)。該方法是有狀態(tài)的，因此不適合并行運行。

想要了解更多開發(fā)技術(shù)，面試教程以及互聯(lián)網(wǎng)公司內(nèi)推，歡迎關(guān)注我的微信公眾號！將會不定期的發(fā)放福利哦~