Java8 新特性之流式數據處理(轉)

時間 2019-11-19

原文原文鏈接

轉自:http://www.javashuo.com/article/p-zpihxovp-ey.html

一. 流式處理簡介

在我接觸到java8流式處理的時候，個人第一感受是流式處理讓集合操做變得簡潔了許多，一般咱們須要多行代碼才能完成的操做，藉助於流式處理能夠在一行中實現。好比咱們但願對一個包含整數的集合中篩選出全部的偶數，並將其封裝成爲一個新的List返回，那麼在java8以前，咱們須要經過以下代碼實現：html

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

經過java8的流式處理，咱們能夠將代碼簡化爲：java

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

先簡單解釋一下上面這行語句的含義，stream()操做將集合轉換成一個流，filter()執行咱們自定義的篩選處理，這裏是經過lambda表達式篩選出全部偶數，最後咱們經過collect()對結果進行封裝處理，並經過Collectors.toList()指定其封裝成爲一個List集合返回。數據庫

由上面的例子能夠看出，java8的流式處理極大的簡化了對於集合的操做，實際上不光是集合，包括數組、文件等，只要是能夠轉換成流，咱們均可以藉助流式處理，相似於咱們寫SQL語句同樣對其進行操做。java8經過內部迭代來實現對流的處理，一個流式處理能夠分爲三個部分：轉換成流、中間操做、終端操做。以下圖：數組

以集合爲例，一個流式處理的操做咱們首先須要調用stream()函數將其轉換成流，而後再調用相應的中間操做達到咱們須要對集合進行的操做，好比篩選、轉換等，最後經過終端操做對前面的結果進行封裝，返回咱們須要的形式。安全

二. 中間操做

咱們定義一個簡單的學生實體類，用於後面的例子演示：多線程

public class Student {

    /** 學號 */
    private long id;

    private String name;

    private int age;

    /** 年級 */
    private int grade;

    /** 專業 */
    private String major;

    /** 學校 */
    private String school;

    // 省略getter和setter
}

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

2.1 過濾

過濾，顧名思義就是按照給定的要求對集合進行篩選知足條件的元素，java8提供的篩選操做包括：filter、distinct、limit、skip。app

filter
在前面的例子中咱們已經演示瞭如何使用filter，其定義爲：Stream<T> filter(Predicate<? super T> predicate)，filter接受一個謂詞Predicate，咱們能夠經過這個謂詞定義篩選條件，在介紹lambda表達式時咱們介紹過Predicate是一個函數式接口，其包含一個test(T t)方法，該方法返回boolean。如今咱們但願從集合students中篩選出全部武漢大學的學生，那麼咱們能夠經過filter來實現，並將篩選操做做爲參數傳遞給filter：框架

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

distinct
distinct操做相似於咱們在寫SQL語句時，添加的DISTINCT關鍵字，用於去重處理，distinct基於Object.equals(Object)實現，回到最開始的例子，假設咱們但願篩選出全部不重複的偶數，那麼能夠添加distinct操做：jvm

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

limit
limit操做也相似於SQL語句中的LIMIT關鍵字，不過相對功能較弱，limit返回包含前n個元素的流，當集合大小小於n時，則返回實際長度，好比下面的例子返回前兩個專業爲土木工程專業的學生：ide

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

說到limit，不得不說起一下另一個流操做：sorted。該操做用於對流中元素進行排序，sorted要求待比較的元素必須實現Comparable接口，若是沒有實現也沒關係，咱們能夠將比較器做爲參數傳遞給sorted(Comparator<? super T> comparator)，好比咱們但願篩選出專業爲土木工程的學生，並按年齡從小到大排序，篩選出年齡最小的兩個學生，那麼能夠實現爲：

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

skip
skip操做與limit操做相反，如同其字面意思同樣，是跳過前n個元素，好比咱們但願找出排序在2以後的土木工程專業的學生，那麼能夠實現爲：

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

經過skip，就會跳過前面兩個元素，返回由後面全部元素構造的流，若是n大於知足條件的集合的長度，則會返回一個空的集合。

2.2 映射

在SQL中，藉助SELECT關鍵字後面添加須要的字段名稱，能夠僅輸出咱們須要的字段數據，而流式處理的映射操做也是實現這一目的，在java8的流式處理中，主要包含兩類映射操做：map和flatMap。

map
舉例說明，假設咱們但願篩選出全部專業爲計算機科學的學生姓名，那麼咱們能夠在filter篩選的基礎之上，經過map將學生實體映射成爲學生姓名字符串，具體實現以下：

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

除了上面這類基礎的map，java8還提供了mapToDouble(ToDoubleFunction<? super T> mapper)，mapToInt(ToIntFunction<? super T> mapper)，mapToLong(ToLongFunction<? super T> mapper)，這些映射分別返回對應類型的流，java8爲這些流設定了一些特殊的操做，好比咱們但願計算全部專業爲計算機科學學生的年齡之和，那麼咱們能夠實現以下：

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

經過將Student按照年齡直接映射爲IntStream，咱們能夠直接調用提供的sum()方法來達到目的，此外使用這些數值流的好處還在於能夠避免jvm裝箱操做所帶來的性能消耗。

flatMap
flatMap與map的區別在於 flatMap是將一個流中的每一個值都轉成一個個流，而後再將這些流扁平化成爲一個流 。舉例說明，假設咱們有一個字符串數組String[] strs = {"java8", "is", "easy", "to", "use"};，咱們但願輸出構成這一數組的全部非重複字符，那麼咱們可能首先會想到以下實現：

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

在執行map操做之後，咱們獲得是一個包含多個字符串（構成一個字符串的字符數組）的流，此時執行distinct操做是基於在這些字符串數組之間的對比，因此達不到咱們但願的目的，此時的輸出爲：

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

distinct只有對於一個包含多個字符的流進行操做才能達到咱們的目的，即對Stream<String>進行操做。此時flatMap就能夠達到咱們的目的：

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

flatMap將由map映射獲得的Stream<String[]>，轉換成由各個字符串數組映射成的流Stream<String>，再將這些小的流扁平化成爲一個由全部字符串構成的大流Steam<String>，從而可以達到咱們的目的。
與map相似，flatMap也提供了針對特定類型的映射操做：flatMapToDouble(Function<? super T,? extends DoubleStream> mapper)，flatMapToInt(Function<? super T,? extends IntStream> mapper)，flatMapToLong(Function<? super T,? extends LongStream> mapper)。

三. 終端操做

終端操做是流式處理的最後一步，咱們能夠在終端操做中實現對流查找、歸約等操做。

3.1 查找

allMatch
allMatch用於檢測是否所有都知足指定的參數行爲，若是所有知足則返回true，例如咱們但願檢測是否全部的學生都已滿18週歲，那麼能夠實現爲：

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

anyMatch
anyMatch則是檢測是否存在一個或多個知足指定的參數行爲，若是知足則返回true，例如咱們但願檢測是否有來自武漢大學的學生，那麼能夠實現爲：

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

noneMathch
noneMatch用於檢測是否不存在知足指定行爲的元素，若是不存在則返回true，例如咱們但願檢測是否不存在專業爲計算機科學的學生，能夠實現以下：

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

findFirst
findFirst用於返回知足條件的第一個元素，好比咱們但願選出專業爲土木工程的排在第一個學生，那麼能夠實現以下：

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

findFirst不攜帶參數，具體的查找條件能夠經過filter設置，此外咱們能夠發現findFirst返回的是一個Optional類型，關於該類型的具體講解能夠參考上一篇：Java8新特性 – Optional類。

findAny
findAny相對於findFirst的區別在於，findAny不必定返回第一個，而是返回任意一個，好比咱們但願返回任意一個專業爲土木工程的學生，能夠實現以下：

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

實際上對於順序流式處理而言，findFirst和findAny返回的結果是同樣的，至於爲何會這樣設計，是由於在下一篇咱們介紹的並行流式處理，當咱們啓用並行流式處理的時候，查找第一個元素每每會有不少限制，若是不是特別需求，在並行流式處理中使用findAny的性能要比findFirst好。

3.2 歸約

前面的例子中咱們大部分都是經過collect(Collectors.toList())對數據封裝返回，如個人目標不是返回一個新的集合，而是但願對通過參數化操做後的集合進行進一步的運算，那麼咱們可用對集合實施歸約操做。java8的流式處理提供了reduce方法來達到這一目的。

前面咱們經過mapToInt將Stream<Student>映射成爲IntStream，並經過IntStream的sum方法求得全部學生的年齡之和，實際上咱們經過歸約操做，也能夠達到這一目的，實現以下：

// 前面例子中的方法
int totalAge = students.stream()
                .filter(student -> "計算機科學".equals(student.getMajor()))
                .mapToInt(Student::getAge).sum();
// 歸約操做
int totalAge = students.stream()
                .filter(student -> "計算機科學".equals(student.getMajor()))
                .map(Student::getAge)
                .reduce(0, (a, b) -> a + b);

// 進一步簡化
int totalAge2 = students.stream()
                .filter(student -> "計算機科學".equals(student.getMajor()))
                .map(Student::getAge)
                .reduce(0, Integer::sum);

// 採用無初始值的重載版本，須要注意返回Optional
Optional<Integer> totalAge = students.stream()
                .filter(student -> "計算機科學".equals(student.getMajor()))
                .map(Student::getAge)
                .reduce(Integer::sum);  // 去掉初始值

3.3 收集

前面利用collect(Collectors.toList())是一個簡單的收集操做，是對處理結果的封裝，對應的還有toSet、toMap，以知足咱們對於結果組織的需求。這些方法均來自於java.util.stream.Collectors，咱們能夠稱之爲收集器。

3.3.1 歸約

收集器也提供了相應的歸約操做，可是與reduce在內部實現上是有區別的，收集器更加適用於可變容器上的歸約操做，這些收集器廣義上均基於Collectors.reducing()實現。

例1：求學生的總人數

long count = students.stream().collect(Collectors.counting());

// 進一步簡化
long count = students.stream().count();

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

// 求最大年齡
Optional<Student> olderStudent = students.stream().collect(Collectors.maxBy((s1, s2) -> s1.getAge() - s2.getAge()));

// 進一步簡化
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));

對應的還有summingLong、summingDouble。

例4：求年齡的平均值

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

對應的還有averagingLong、averagingDouble。

例5：一次性獲得元素個數、總和、均值、最大值、最小值

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

輸出：

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

對應的還有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 分組

在數據庫操做中，咱們能夠經過GROUP BY關鍵字對查詢到的數據進行分組，java8的流式處理也爲咱們提供了這樣的功能Collectors.groupingBy來操做集合。好比咱們能夠按學校對上面的學生進行分組：

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

groupingBy接收一個分類器Function<? super T, ? extends K> classifier，咱們能夠自定義分類器來實現須要的分類效果。

上面演示的是一級分組，咱們還能夠定義多個分類器實現 多級分組，好比咱們但願在按學校分組的基礎之上再按照專業進行分組，實現以下：

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

實際上在groupingBy的第二個參數不是隻能傳遞groupingBy，還能夠傳遞任意Collector類型，好比咱們能夠傳遞一個Collector.counting，用以統計每一個組的個數：

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

若是咱們不添加第二個參數，則編譯器會默認幫咱們添加一個Collectors.toList()。

3.3.3 分區

分區能夠看作是分組的一種特殊狀況，在分區中key只有兩種狀況：true或false，目的是將待分區集合按照條件一分爲二，java8的流式處理利用ollectors.partitioningBy()方法實現分區，該方法接收一個謂詞，例如咱們但願將學生分爲武大學生和非武大學生，那麼能夠實現以下：

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

分區相對分組的優點在於，咱們能夠同時獲得兩類結果，在一些應用場景下能夠一步獲得咱們須要的全部結果，好比將數組分爲奇數和偶數。

以上介紹的全部收集器均實現自接口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();

}

咱們也能夠實現該接口來定義本身的收集器，此處再也不展開。

四. 並行流式數據處理

流式處理中的不少都適合採用 分而治之 的思想，從而在處理集合較大時，極大的提升代碼的性能，java8的設計者也看到了這一點，因此提供了 並行流式處理。上面的例子中咱們都是調用stream()方法來啓動流式處理，java8還提供了parallelStream()來啓動並行流式處理，parallelStream()本質上基於java7的Fork-Join框架實現，其默認的線程數爲宿主機的內核數。

啓動並行流式處理雖然簡單，只須要將stream()替換成parallelStream()便可，但既然是並行，就會涉及到多線程安全問題，因此在啓用以前要先確認並行是否值得（並行的效率不必定高於順序執行），另外就是要保證線程安全。此兩項沒法保證，那麼並行毫無心義，畢竟結果比速度更加劇要，之後有時間再來詳細分析一下並行流式數據處理的具體實現和最佳實踐。

做者：深藍至尊本篇文章是博主原創做品，嚴禁沒有備註的轉載，COPY.