Abstract: Java8’s streaming data processing greatly simplifies our operations on structures such as collections and arrays, allowing us to operate with functional thinking. This article will explore the basic use of Java8’s streaming data processing.

Java8 new feature series

Java8 new feature (1) - lambda expression

Java8 new feature (2) - Optional class

Java8 new feature (3) - Streaming data processing

Java8 new feature (4) - Default interface method

To be determined

1. Introduction to streaming processing

?When I came into contact with Java8 streaming processing, my first feeling was that streaming processing makes collection operations much simpler. Usually we need more Operations that require one line of code can be accomplished in one line with the help of streaming. For example, if we want to filter out all even numbers from a collection containing integers and encapsulate them into a new List for return, then before java8, we need to implement it through the following code:

List<Integer> evens = new ArrayList<>();for (final Integer num : nums) {    if (num % 2 == 0) {
        evens.add(num);
    }
}

Through the streaming processing of java8, we The code can be simplified as:

List<Integer> evens = nums.stream().filter(num -> num % 2 == 0).collect(Collectors.toList());

Let’s briefly explain the meaning of the above line of statement. The stream() operation converts the collection into a stream, and filter() performs our custom filtering process. Here, we filter out through lambda expressions. For all even numbers, we finally encapsulate the result through collect(), and specify its encapsulation into a List collection and return it through Collectors.toList().
As can be seen from the above example, Java8's streaming processing greatly simplifies the operation of collections. In fact, it is not just collections, including arrays, files, etc., as long as it can be converted into a stream, we can use streaming processing , operate on it similar to how we write SQL statements. Java8 implements stream processing through internal iteration. A stream processing can be divided into three parts: conversion to stream, intermediate operation, and terminal operation. As shown below:

Taking a collection as an example, for a streaming operation, we first need to call the stream() function to convert it into a stream, and then call the corresponding intermediate operations to achieve the operations we need to perform on the collection, such as Filtering, conversion, etc., and finally encapsulating the previous results through terminal operations and returning the form we need.

2. Intermediate operations

We define a simple student entity class for later example demonstrations:

public class Student {    /** 學(xué)號 */
    private long id;    private String name;    private int age;    /** 年級 */
    private int grade;    /** 專業(yè) */
    private String major;    /** 學(xué)校 */
    private String school;    // 省略getter和setter}

// 初始化List<Student> students = new ArrayList<Student>() {
    {
        add(new Student(20160001, "孔明", 20, 1, "土木工程", "武漢大學(xué)"));
        add(new Student(20160002, "伯約", 21, 2, "信息安全", "武漢大學(xué)"));
        add(new Student(20160003, "玄德", 22, 3, "經(jīng)濟管理", "武漢大學(xué)"));
        add(new Student(20160004, "云長", 21, 2, "信息安全", "武漢大學(xué)"));
        add(new Student(20161001, "翼德", 21, 2, "機械與自動化", "華中科技大學(xué)"));
        add(new Student(20161002, "元直", 23, 4, "土木工程", "華中科技大學(xué)"));
        add(new Student(20161003, "奉孝", 23, 4, "計算機科學(xué)", "華中科技大學(xué)"));
        add(new Student(20162001, "仲謀", 22, 3, "土木工程", "浙江大學(xué)"));
        add(new Student(20162002, "魯肅", 23, 4, "計算機科學(xué)", "浙江大學(xué)"));
        add(new Student(20163001, "丁奉", 24, 5, "土木工程", "南京大學(xué)"));
    }
};

2.1 Filtering

Filtering, as the name suggests, is to filter the elements that meet the conditions in the collection according to the given requirements, java8 The filtering operations provided include: filter, distinct, limit, and skip.

filter

? In the previous example we have demonstrated how to use filter, which is defined as: Stream filter(Predicate predicate). Filter accepts a predicate Predicate. We can define filtering conditions through this predicate. Introducing lambda expression We introduced that Predicate is a functional interface, which contains a test(T t) method, which returns boolean. Now we want to filter out all Wuhan University students from the students collection, then we can do it through filter, and pass the filtering operation as a parameter to the filter:

List<Student> whuStudents = students.stream()
                                    .filter(student -> "武漢大學(xué)".equals(student.getSchool()))
                                    .collect(Collectors.toList());

distinct

? The distinct operation is similar to when we write a SQL statement, The added DISTINCT keyword is used for deduplication processing. Distinct is implemented based on Object.equals(Object). Going back to the original example, assuming we want to filter out all non-repeating even numbers, we can add the distinct operation:

List<Integer> evens = nums.stream()
                        .filter(num -> num % 2 == 0).distinct()
                        .collect(Collectors.toList());

limit

?The limit operation is also similar to the LIMIT keyword in the SQL statement, but its function is relatively weak. Limit returns a stream containing the first n elements. When the set size is less than n, the actual length is returned. For example, the following example returns the first two A student majoring in civil engineering:

List<Student> civilStudents = students.stream()
                                    .filter(student -> "土木工程".equals(student.getMajor())).limit(2)
                                    .collect(Collectors.toList());

Speaking of limit, I have to mention another stream operation: sorted. This operation is used to sort the elements in the stream. sorted requires that the elements to be compared must implement the Comparable interface. It doesn’t matter if it is not implemented. We can pass the comparator as a parameter to sorted (Comparator comparator). For example, we want to filter out the majors Civil engineering students, and sorted by age from small to large, and filter out the two youngest students, then it can be implemented as:

List<Student> sortedCivilStudents = students.stream()
                                            .filter(student -> "土木工程".equals(student.getMajor())).sorted((s1, s2) -> s1.getAge() - s2.getAge())
                                            .limit(2)
                                            .collect(Collectors.toList());

skip

? The skip operation is the opposite of the limit operation. Just like its literal meaning, it skips the previous ones. n elements, for example, if we want to find the civil engineering students ranked after 2, then it can be implemented as:

List<Student> civilStudents = students.stream()
                                    .filter(student -> "土木工程".equals(student.getMajor()))
                                    .skip(2)
                                    .collect(Collectors.toList());

通過skip，就會跳過前面兩個元素，返回由后面所有元素構(gòu)造的流，如果n大于滿足條件的集合的長度，則會返回一個空的集合。

2.2 映射

??在SQL中，借助SELECT關(guān)鍵字后面添加需要的字段名稱，可以僅輸出我們需要的字段數(shù)據(jù)，而流式處理的映射操作也是實現(xiàn)這一目的，在java8的流式處理中，主要包含兩類映射操作：map和flatMap。

map

??舉例說明，假設(shè)我們希望篩選出所有專業(yè)為計算機科學(xué)的學(xué)生姓名，那么我們可以在filter篩選的基礎(chǔ)之上，通過map將學(xué)生實體映射成為學(xué)生姓名字符串，具體實現(xiàn)如下：

List<String> names = students.stream()
                            .filter(student -> "計算機科學(xué)".equals(student.getMajor()))
                            .map(Student::getName).collect(Collectors.toList());

除了上面這類基礎(chǔ)的map，java8還提供了mapToDouble(ToDoubleFunction mapper)，mapToInt(ToIntFunction mapper)，mapToLong(ToLongFunction mapper)，這些映射分別返回對應(yīng)類型的流，java8為這些流設(shè)定了一些特殊的操作，比如我們希望計算所有專業(yè)為計算機科學(xué)學(xué)生的年齡之和，那么我們可以實現(xiàn)如下：

int totalAge = students.stream()
                    .filter(student -> "計算機科學(xué)".equals(student.getMajor()))
                    .mapToInt(Student::getAge).sum();

通過將Student按照年齡直接映射為IntStream，我們可以直接調(diào)用提供的sum()方法來達(dá)到目的，此外使用這些數(shù)值流的好處還在于可以避免jvm裝箱操作所帶來的性能消耗。

flatMap

??flatMap與map的區(qū)別在于 flatMap是將一個流中的每個值都轉(zhuǎn)成一個個流，然后再將這些流扁平化成為一個流 。舉例說明，假設(shè)我們有一個字符串?dāng)?shù)組String[] strs = {"java8", "is", "easy", "to", "use"};，我們希望輸出構(gòu)成這一數(shù)組的所有非重復(fù)字符，那么我們可能首先會想到如下實現(xiàn)：

List<String[]> distinctStrs = Arrays.stream(strs)
                                .map(str -> str.split(""))  // 映射成為Stream<String[]>
                                .distinct()
                                .collect(Collectors.toList());

在執(zhí)行map操作以后，我們得到是一個包含多個字符串（構(gòu)成一個字符串的字符數(shù)組）的流，此時執(zhí)行distinct操作是基于在這些字符串?dāng)?shù)組之間的對比，所以達(dá)不到我們希望的目的，此時的輸出為：

[j, a, v, a, 8]
[i, s]
[e, a, s, y]
[t, o]
[u, s, e]

distinct只有對于一個包含多個字符的流進(jìn)行操作才能達(dá)到我們的目的，即對Stream進(jìn)行操作。此時flatMap就可以達(dá)到我們的目的：

List<String> distinctStrs = Arrays.stream(strs)
                                .map(str -> str.split(""))  // 映射成為Stream<String[]>
                                .flatMap(Arrays::stream)  // 扁平化為Stream<String>
                                .distinct()
                                .collect(Collectors.toList());

flatMap將由map映射得到的Stream，轉(zhuǎn)換成由各個字符串?dāng)?shù)組映射成的流Stream，再將這些小的流扁平化成為一個由所有字符串構(gòu)成的大流Steam，從而能夠達(dá)到我們的目的。
??與map類似，flatMap也提供了針對特定類型的映射操作：flatMapToDouble(Function mapper)，flatMapToInt(Function mapper)，flatMapToLong(Function mapper)。

三. 終端操作

??終端操作是流式處理的最后一步，我們可以在終端操作中實現(xiàn)對流查找、歸約等操作。

3.1 查找

allMatch

??allMatch用于檢測是否全部都滿足指定的參數(shù)行為，如果全部滿足則返回true，例如我們希望檢測是否所有的學(xué)生都已滿18周歲，那么可以實現(xiàn)為：

boolean isAdult = students.stream().allMatch(student -> student.getAge() >= 18);

anyMatch

??anyMatch則是檢測是否存在一個或多個滿足指定的參數(shù)行為，如果滿足則返回true，例如我們希望檢測是否有來自武漢大學(xué)的學(xué)生，那么可以實現(xiàn)為：

boolean hasWhu = students.stream().anyMatch(student -> "武漢大學(xué)".equals(student.getSchool()));

noneMathch

??noneMatch用于檢測是否不存在滿足指定行為的元素，如果不存在則返回true，例如我們希望檢測是否不存在專業(yè)為計算機科學(xué)的學(xué)生，可以實現(xiàn)如下：

boolean noneCs = students.stream().noneMatch(student -> "計算機科學(xué)".equals(student.getMajor()));

findFirst

??findFirst用于返回滿足條件的第一個元素，比如我們希望選出專業(yè)為土木工程的排在第一個學(xué)生，那么可以實現(xiàn)如下：

Optional

Optional<Student> optStu = students.stream().filter(student -> "土木工程".equals(student.getMajor())).findFirs

實際上對于順序流式處理而言，findFirst和findAny返回的結(jié)果是一樣的，至于為什么會這樣設(shè)計，是因為在下一篇我們介紹的 并行流式處理，當(dāng)我們啟用并行流式處理的時候，查找第一個元素往往會有很多限制，如果不是特別需求，在并行流式處理中使用findAny的性能要比findFirst好。

3.2 歸約

??前面的例子中我們大部分都是通過collect(Collectors.toList())對數(shù)據(jù)封裝返回，如我的目標(biāo)不是返回一個新的集合，而是希望對經(jīng)過參數(shù)化操作后的集合進(jìn)行進(jìn)一步的運算，那么我們可用對集合實施歸約操作。java8的流式處理提供了reduce方法來達(dá)到這一目的。
??前面我們通過mapToInt將Stream映射成為IntStream，并通過IntStream的sum方法求得所有學(xué)生的年齡之和，實際上我們通過歸約操作，也可以達(dá)到這一目的，實現(xiàn)如下：

// 前面例子中的方法int totalAge = students.stream()
                .filter(student -> "計算機科學(xué)".equals(student.getMajor()))
                .mapToInt(Student::getAge).sum();// 歸約操作int totalAge = students.stream()
                .filter(student -> "計算機科學(xué)".equals(student.getMajor()))
                .map(Student::getAge)
                .reduce(0, (a, b) -> a + b);// 進(jìn)一步簡化int totalAge2 = students.stream()
                .filter(student -> "計算機科學(xué)".equals(student.getMajor()))
                .map(Student::getAge)
                .reduce(0, Integer::sum);// 采用無初始值的重載版本，需要注意返回OptionalOptional<Integer> totalAge = students.stream()
                .filter(student -> "計算機科學(xué)".equals(student.getMajor()))
                .map(Student::getAge)
                .reduce(Integer::sum);  // 去掉初始值

3.3 收集

??前面利用collect(Collectors.toList())是一個簡單的收集操作，是對處理結(jié)果的封裝，對應(yīng)的還有toSet、toMap，以滿足我們對于結(jié)果組織的需求。這些方法均來自于java.util.stream.Collectors，我們可以稱之為收集器。

3.3.1 歸約

??收集器也提供了相應(yīng)的歸約操作，但是與reduce在內(nèi)部實現(xiàn)上是有區(qū)別的，收集器更加適用于可變?nèi)萜魃系臍w約操作，這些收集器廣義上均基于Collectors.reducing()實現(xiàn)。

例1：求學(xué)生的總?cè)藬?shù)

long count = students.stream().collect(Collectors.counting());// 進(jìn)一步簡化long count = students.stream().count();

例2：求年齡的最大值和最小值

// 求最大年齡Optional<Student> olderStudent = students.stream().collect(Collectors.maxBy((s1, s2) -> s1.getAge() - s2.getAge()));// 進(jìn)一步簡化Optional<Student> olderStudent2 = students.stream().collect(Collectors.maxBy(Comparator.comparing(Student::getAge)));// 求最小年齡Optional<Student> olderStudent3 = students.stream().collect(Collectors.minBy(Comparator.comparing(Student::getAge)));

例3：求年齡總和

int totalAge4 = students.stream().collect(Collectors.summingInt(Student::getAge));

對應(yīng)的還有summingLong、summingDouble。

例4：求年齡的平均值

double avgAge = students.stream().collect(Collectors.averagingInt(Student::getAge));

對應(yīng)的還有averagingLong、averagingDouble。

例5：一次性得到元素個數(shù)、總和、均值、最大值、最小值

IntSummaryStatistics statistics = students.stream().collect(Collectors.summarizingInt(Student::getAge));

輸出：

IntSummaryStatistics{count=10, sum=220, min=20, average=22.000000, max=24}

對應(yīng)的還有summarizingLong、summarizingDouble。

例6：字符串拼接

String names = students.stream().map(Student::getName).collect(Collectors.joining());
// 輸出：孔明伯約玄德云長翼德元直奉孝仲謀魯肅丁奉
String names = students.stream().map(Student::getName).collect(Collectors.joining(", "));
// 輸出：孔明, 伯約, 玄德, 云長, 翼德, 元直, 奉孝, 仲謀, 魯肅, 丁奉

3.3.2 分組

??在數(shù)據(jù)庫操作中，我們可以通過GROUP BY關(guān)鍵字對查詢到的數(shù)據(jù)進(jìn)行分組，java8的流式處理也為我們提供了這樣的功能Collectors.groupingBy來操作集合。比如我們可以按學(xué)校對上面的學(xué)生進(jìn)行分組：

Map<String, List<Student>> groups = students.stream().collect(Collectors.groupingBy(Student::getSchool));

groupingBy接收一個分類器Function classifier，我們可以自定義分類器來實現(xiàn)需要的分類效果。
??上面演示的是一級分組，我們還可以定義多個分類器實現(xiàn) 多級分組，比如我們希望在按學(xué)校分組的基礎(chǔ)之上再按照專業(yè)進(jìn)行分組，實現(xiàn)如下：

Map<String, Map<String, List<Student>>> groups2 = students.stream().collect(
                Collectors.groupingBy(Student::getSchool,  // 一級分組，按學(xué)校
                Collectors.groupingBy(Student::getMajor)));  // 二級分組，按專業(yè)

實際上在groupingBy的第二個參數(shù)不是只能傳遞groupingBy，還可以傳遞任意Collector類型，比如我們可以傳遞一個Collector.counting，用以統(tǒng)計每個組的個數(shù)：

Map<String, Long> groups = students.stream().collect(Collectors.groupingBy(Student::getSchool, Collectors.counting()));

如果我們不添加第二個參數(shù)，則編譯器會默認(rèn)幫我們添加一個Collectors.toList()。

3.3.3 分區(qū)

??分區(qū)可以看做是分組的一種特殊情況，在分區(qū)中key只有兩種情況：true或false，目的是將待分區(qū)集合按照條件一分為二，java8的流式處理利用ollectors.partitioningBy()方法實現(xiàn)分區(qū)，該方法接收一個謂詞，例如我們希望將學(xué)生分為武大學(xué)生和非武大學(xué)生，那么可以實現(xiàn)如下：

Map<Boolean, List<Student>> partition = students.stream().collect(Collectors.partitioningBy(student -> "武漢大學(xué)".equals(student.getSchool())));

分區(qū)相對分組的優(yōu)勢在于，我們可以同時得到兩類結(jié)果，在一些應(yīng)用場景下可以一步得到我們需要的所有結(jié)果，比如將數(shù)組分為奇數(shù)和偶數(shù)。

??以上介紹的所有收集器均實現(xiàn)自接口java.util.stream.Collector，該接口的定義如下：

public interface Collector<T, A, R> {    /**
     * A function that creates and returns a new mutable result container.
     *
     * @return a function which returns a new, mutable result container
     */
    Supplier<A> supplier();    /**
     * A function that folds a value into a mutable result container.
     *
     * @return a function which folds a value into a mutable result container
     */
    BiConsumer<A, T> accumulator();    /**
     * A function that accepts two partial results and merges them.  The
     * combiner function may fold state from one argument into the other and
     * return that, or may return a new result container.
     *
     * @return a function which combines two partial results into a combined
     * result
     */
    BinaryOperator<A> combiner();    /**
     * Perform the final transformation from the intermediate accumulation type
     * {@code A} to the final result type {@code R}.
     *
     * <p>If the characteristic {@code IDENTITY_TRANSFORM} is
     * set, this function may be presumed to be an identity transform with an
     * unchecked cast from {@code A} to {@code R}.
     *
     * @return a function which transforms the intermediate result to the final
     * result
     */
    Function<A, R> finisher();    /**
     * Returns a {@code Set} of {@code Collector.Characteristics} indicating
     * the characteristics of this Collector.  This set should be immutable.
     *
     * @return an immutable set of collector characteristics
     */
    Set<Characteristics> characteristics();

}

我們也可以實現(xiàn)該接口來定義自己的收集器，此處不再展開。

四. 并行流式數(shù)據(jù)處理

??流式處理中的很多都適合采用?分而治之?的思想，從而在處理集合較大時，極大的提高代碼的性能，java8的設(shè)計者也看到了這一點，所以提供了?并行流式處理。上面的例子中我們都是調(diào)用stream()方法來啟動流式處理，java8還提供了parallelStream()來啟動并行流式處理，parallelStream()本質(zhì)上基于java7的Fork-Join框架實現(xiàn)，其默認(rèn)的線程數(shù)為宿主機的內(nèi)核數(shù)。
??啟動并行流式處理雖然簡單，只需要將stream()替換成parallelStream()即可，但既然是并行，就會涉及到多線程安全問題，所以在啟用之前要先確認(rèn)并行是否值得（并行的效率不一定高于順序執(zhí)行），另外就是要保證線程安全。此兩項無法保證，那么并行毫無意義，畢竟結(jié)果比速度更加重要，以后有時間再來詳細(xì)分析一下并行流式數(shù)據(jù)處理的具體實現(xiàn)和最佳實踐。

? 著作權(quán)歸作者所有