簡介

? ? Java 8里引入的另外一個(gè)重要特性就是stream api?；\統(tǒng)的來說，它這種特性的引入可以方便我們以一種更加聲明式的方式來寫代碼，更加便利了一些函數(shù)式編程方法的使用。同時(shí)，它也使得我們可以充分利用系統(tǒng)的并行能力而不用自己手工的去做很多底層的工作。當(dāng)然，里面最讓人印象深刻的也許是一種類似于流式編程的概念。?

?

流水線(pipeline)

? ? 在以前一些linux腳本命令中經(jīng)常會(huì)接觸到的一個(gè)概念就是pipeline，它其實(shí)體現(xiàn)出來了一個(gè)很好的程序設(shè)計(jì)哲學(xué)，就是我們應(yīng)該設(shè)計(jì)很多小而且職責(zé)單一的模塊。每個(gè)模塊只專注于做一件事情，然后它們之間通過一種流水線的方式將它們連接起來。我們看一個(gè)典型的命令：?

cat file1 file2 | tr "[A-Z]" "[a-z]" | sort | tail -3

? ? 上面這部分的命令表示從file1, file2兩個(gè)文件里讀內(nèi)容，然后將里面的大寫字母轉(zhuǎn)換成小寫字母，然后再排序。最后取排序后的最后3個(gè)字符。

? ? 我們這里其實(shí)不是關(guān)心這個(gè)命令做了什么，而是這些命令它們執(zhí)行的方式。實(shí)際上，在linux里面，上述的幾個(gè)命令它們完全是并發(fā)執(zhí)行的，前面的cat命令可能是讀取了一部分文件的內(nèi)容經(jīng)由tr命令替換字符后，再由sort命令排序。它們的執(zhí)行過程如下圖所示：

? ? ?上述的執(zhí)行過程類似于一個(gè)工廠里的生產(chǎn)流水線，在每個(gè)生產(chǎn)的步驟里，它不是等前面一個(gè)步驟要生產(chǎn)的所有東西都完成才做下一步，而是前面做完一部分就馬上傳遞給后面一個(gè)部分。這樣才能實(shí)現(xiàn)所有步驟的并發(fā)工作。如果熟悉python的同學(xué)，也許會(huì)聯(lián)想到里面的generator的功能，它的功能也是類似的。

? ? 那么，上述的這種流水線式的編程方式有什么好處呢？除了前面提到的它可以使得我們充分利用計(jì)算機(jī)的并發(fā)能力，還能夠處理一些數(shù)據(jù)量很大的場(chǎng)景。因?yàn)樗皇撬械臄?shù)據(jù)都要一次性的放到內(nèi)存里來處理。另外，它的每個(gè)步驟如果定義好之后，確實(shí)可以結(jié)合前面函數(shù)式編程的討論得到一個(gè)很好的應(yīng)用。

? ? 現(xiàn)在，java 8里面引入的stream特性，就是給我們帶來了上述的好處。我們來詳細(xì)分析一下。

?

示例對(duì)比

? ? 假設(shè)我們有一個(gè)如下類：

import java.util.*;public class Dish {private final String name;private final boolean vegetarian;private final int calories;private final Type type;public Dish(String name, boolean vegetarian, int calories, Type type) {this.name = name;this.vegetarian = vegetarian;this.calories = calories;this.type = type;}public String getName() {return name;}public boolean isVegetarian() {return vegetarian;}public int getCalories() {return calories;}public Type getType() {return type;}public enum Type { MEAT, FISH, OTHER }@Overridepublic String toString() {return name;}public static final List<Dish> menu =Arrays.asList( new Dish("pork", false, 800, Dish.Type.MEAT),new Dish("beef", false, 700, Dish.Type.MEAT),new Dish("chicken", false, 400, Dish.Type.MEAT),new Dish("french fries", true, 530, Dish.Type.OTHER),new Dish("rice", true, 350, Dish.Type.OTHER),new Dish("season fruit", true, 120, Dish.Type.OTHER),new Dish("pizza", true, 550, Dish.Type.OTHER),new Dish("prawns", false, 400, Dish.Type.FISH),new Dish("salmon", false, 450, Dish.Type.FISH)); }

? ? 這個(gè)示例稍微有點(diǎn)長，主要是定義了一個(gè)Dish對(duì)象，然后初始化了一個(gè)Dish的list。?

? ? 現(xiàn)在假設(shè)我們需要做一些如下的操作，首先獲取列表里卡路里小于400的元素，然后再根據(jù)卡路里的數(shù)值進(jìn)行排序，最后我們?cè)俜祷剡@些排序后的元素的名字。如果按照我們通常的理解，會(huì)做一個(gè)如下的實(shí)現(xiàn)：

List<Dish> lowCaloricDishes = new ArrayList<>(); for(Dish d: menu) {if(d.getCalories() < 400) {lowCaloricDishes.add(d);} } Collections.sort(lowCaloricDishes, new Comparator<Dish>() {public int compare(Dish d1, Dish d2) {return Integer.compare(d1.getCalories(), d2.getCalories()); } }); List<String> lowCaloricDishesName = new ArrayList<>(); for(Dish d: lowCaloricDishes) {lowCaloricDishesName.add(d.getName()); }

?? ?上面這部分的代碼看起來很中規(guī)中矩，當(dāng)然，也顯得有點(diǎn)啰嗦。具體它的特點(diǎn)以及與后面的代碼對(duì)比會(huì)在后面詳細(xì)說。如果我們用stream api來實(shí)現(xiàn)上述的邏輯該怎么做呢？

?

import static java.util.Comparator.comparing; import static java.til.stream.Collectors.toList;List<String> lowCaloricDishesName = menu.stream().filter(d -> d.getCalories < 400).sorted(comparing(Dish::getCalories)).map(Dish::getName).collect(toList());

? ? ?現(xiàn)在我們來詳細(xì)比較一下兩種寫法上的差別。在第一種寫法上，我們需要過濾數(shù)據(jù)元素的時(shí)候需要使用一個(gè)臨時(shí)的list來保存過濾后的結(jié)果，然后再將過濾后的元素排序。因?yàn)槲覀冏詈笮枰氖且粋€(gè)排序后元素的名字列表，于是沒辦法，又要?jiǎng)?chuàng)建一個(gè)list，將里面的元素一個(gè)個(gè)的獲取出來再填充到這個(gè)list里。所以綜合來說，這種方法需要?jiǎng)?chuàng)建大量臨時(shí)的列表。這樣不但使得程序變得冗長難懂，而且創(chuàng)建的這些臨時(shí)的列表也增加了程序垃圾回收的壓力。

? ? 我們?cè)倏磗tream api的實(shí)現(xiàn)方式。上述代碼的實(shí)現(xiàn)更加是聲明式的，它的處理流程更加像一個(gè)流水線的方式。我們首先利用filter方法來過濾元素，然后調(diào)用sorted方法來排序，最后用map方法來轉(zhuǎn)換提取的元素。這種寫法不僅更加簡潔而且更加高效。關(guān)于這些具體方法的意思我們?cè)诤罄m(xù)部分詳細(xì)討論。

?

stream定義

? ? 從前面使用手法上來看，stream的使用像是一個(gè)流水線。在這個(gè)流水線里，它好像有一個(gè)自動(dòng)推進(jìn)的流程，我們使用者只需要指定對(duì)它的各種轉(zhuǎn)換操作就可以了。從更嚴(yán)格的意義來說，stream是一組定義的計(jì)算序列，這種結(jié)構(gòu)將一系列的操作給串聯(lián)起來。所以如果熟悉設(shè)計(jì)模式的人會(huì)覺得這就像是一個(gè)chain of responsibility模式。當(dāng)然，從函數(shù)式編程的理論角度來說，它表示的是一個(gè)叫monad的結(jié)構(gòu)。

? ? 因此，從定義的角度來說，stream定義的并不是一個(gè)普通意義上的數(shù)據(jù)流，它實(shí)際上是一個(gè)計(jì)算流，表示一組計(jì)算的順序。它有一些典型的特性，比如內(nèi)循環(huán)(internal iteration), 惰性計(jì)算(laziness)等。我們結(jié)合它們和集合類的比較來一起討論。

?

內(nèi)迭代和外迭代(internal iteration vs external iteration)

? ? 在前面示例代碼里，我們已經(jīng)比對(duì)過兩種實(shí)現(xiàn)方法，對(duì)于第一種方法來說，它需要顯式的定義一個(gè)循環(huán)迭代的過程。比如：

for(Dish d: menu) {if(d.getCalories() < 400) {lowCaloricDishes.add(d);} }

? ? ?這部分代碼的本質(zhì)是集合實(shí)現(xiàn)了一個(gè)iterable的接口，然后在這個(gè)循環(huán)里調(diào)用iterator()方法這樣依次的遍歷集合里的元素。這種方式實(shí)現(xiàn)的代碼有如下幾個(gè)問題：

1. for循環(huán)本身就是串行的過程，所有集合里元素處理的順序必須按照定義好的順序來處理。

2. 因?yàn)檫@種循環(huán)是由開發(fā)人員來寫的，而不是本身庫內(nèi)部定義的，這樣系統(tǒng)比較難做一些內(nèi)在的優(yōu)化，比如數(shù)據(jù)的重排序，潛在并行性的利用等。

? ? 尤其是牽涉到大量數(shù)據(jù)和性能的時(shí)候，如果有更加好的方式來優(yōu)雅的處理程序邏輯將更加受到歡迎。

? ? 與前面對(duì)應(yīng)的是另外一種遍歷方式，稱為內(nèi)部迭代。和上述代碼對(duì)應(yīng)的一種實(shí)現(xiàn)如下：

menu.stream().filter(d -> d.getCalories < 400)

? ? ?從語法上看起來它只是一個(gè)很小的變化，但是它的實(shí)際實(shí)現(xiàn)卻是差別很大的。因?yàn)檫@里的代碼并沒有顯式的定義循環(huán)處理的過程，真正迭代處理的過程相當(dāng)于交給類庫來處理了。類庫的實(shí)現(xiàn)可以潛在的利用一些優(yōu)化的手段來使得程序的執(zhí)行性能更加高效。所以一旦看到stream的時(shí)候，對(duì)它執(zhí)行運(yùn)算時(shí)就好像已經(jīng)在一個(gè)生產(chǎn)線的傳送帶上了。所有需要做的事情就是將一些具體的操作傳遞給這個(gè)流水線。

? ? 前面這種方式的實(shí)現(xiàn)實(shí)際上將要做什么和怎么做是混在一起的。比如說我需要過濾出來所有卡路里小于400的菜，這里就需要循環(huán)遍歷所有的菜列表。而后面的這種方式更像是一個(gè)聲明，只是說我需要過濾某個(gè)東西。而這個(gè)東西的條件就是一個(gè)lambda表達(dá)式，至于它的過濾是怎么實(shí)現(xiàn)的我們可以不用去關(guān)心了。 這樣整個(gè)業(yè)務(wù)邏輯的代碼實(shí)現(xiàn)也更加清晰簡練。

?

stream工作方式

不可變性

? ? 基于前面的示例，我們可能有若干個(gè)疑問，因?yàn)榍懊姘凑諅鹘y(tǒng)的方式來實(shí)現(xiàn)的功能需要用到臨時(shí)的列表，必然要修改一些元素的屬性。那么在stream里面，我們調(diào)用的那些處理方法它會(huì)不會(huì)修改原有stream數(shù)據(jù)源的值呢？我們看如下的代碼：

?

List<String> myList = new ArrayList<>(); myList.add("a1"); myList.add("a2"); myList.add("b1"); myList.add("c2"); myList.add("c1"); myList.stream().filter(s -> s.startsWith("a")).map(String::toUpperCase).sorted().forEach(System.out::println); System.out.println(myList);

? ?它的輸出如下：

A1 A2 [a1, a2, b1, c2, c1]

? ? 上述代碼里的filter, map等方法并沒有修改stream里源的內(nèi)容。它僅僅是根據(jù)當(dāng)前的轉(zhuǎn)換操作新建一個(gè)元素。這種思路恰恰也是和copy on write的數(shù)據(jù)結(jié)構(gòu)暗合的。而且它對(duì)于以后的并發(fā)處理也是有巨大的好處。

?

不可重復(fù)使用

? ? stream還有一個(gè)典型的特征就是它不能被重復(fù)使用，比如說我們嘗試如下的代碼：

Stream<String> stream = myList.stream();stream.anyMatch(s -> true); stream.anyMatch(s -> true);

? ?在編譯的時(shí)候沒有問題，而運(yùn)行的時(shí)候?qū)⒊霈F(xiàn)如下的錯(cuò)誤：

?

Exception in thread "main" java.lang.IllegalStateException: stream has already been operated upon or closedat java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:229)at java.util.stream.ReferencePipeline.anyMatch(ReferencePipeline.java:449)at Sample.main(Sample.java:20)

? ? 因此，凡是我們使用的stream它就相當(dāng)于一次性的用品，用完之后就會(huì)被close了。如果我們需要再利用stream進(jìn)一步的操作需要重新聲明一個(gè)新的stream。

?

兩種運(yùn)算

? ? 在前面的代碼里還要一個(gè)需要我們深入了解的地方就是，我們能夠?qū)σ粋€(gè)stream做哪些操作呢？像前面的filter, map, forEach, collect等。它們有什么作用呢？

? ? 在stream里，主要有兩種運(yùn)算，一種叫中間運(yùn)算(intermediate)，還要一種是終止運(yùn)算(terminal)。比如前面的filter, map運(yùn)算。filter運(yùn)算僅僅過濾stream里的元素，但是返回的依然是一個(gè)Stream<String>類型。同樣，map操作也僅僅實(shí)現(xiàn)一個(gè)元素的轉(zhuǎn)換。如果我們有一些類型轉(zhuǎn)換的話，實(shí)際上也只是將一種類型參數(shù)的Stream轉(zhuǎn)換成另外一種Stream。而終止運(yùn)算比如前面的collect，它將一個(gè)Stream又轉(zhuǎn)換成了一個(gè)List，類似的它還要toSet等方法。這些方法使得stream的處理終止。所以我們稱之為終止運(yùn)算方法。關(guān)于intermediate和terminal方法的詳細(xì)介紹可以參考Stream的官方文檔，如下鏈接。

??

惰性計(jì)算(laziness)

? ? stream里還要一個(gè)比較典型的特性就是惰性計(jì)算。像前面stream里的一些典型運(yùn)算filter, mapping。它們可以通過急性求值的方式來實(shí)現(xiàn)。以filter方法為例，也就是說在方法返回前，急性求值就需要完成對(duì)所有數(shù)據(jù)元素的過濾。而惰性計(jì)算則是當(dāng)需要的時(shí)候才進(jìn)行過濾運(yùn)算。在實(shí)際應(yīng)用中，惰性計(jì)算的方式更加有優(yōu)勢(shì)一些。因?yàn)槲覀儗⑺土魉€后續(xù)的一些操作結(jié)合在一起來運(yùn)算，而不用多次遍歷數(shù)據(jù)。在某些場(chǎng)景里，比如說我們需要遍歷一個(gè)非常大的集合去尋找第一個(gè)匹配某種條件的數(shù)據(jù)，我們完全可以在找到第一個(gè)匹配的元素時(shí)就返回，而不用真正去完整遍歷整個(gè)集合。這種特性尤其在數(shù)據(jù)集合有無限個(gè)長度的情況下用處比較明顯。

? ? 我們來看一個(gè)如下的示例：

Stream.of("d2", "a2", "b1", "b3", "c").filter(s -> {System.out.println("filter: " + s);return true;});

? ? 上述的stream操作里只有一個(gè)filter操作，相當(dāng)于只是做了一個(gè)stream轉(zhuǎn)換成另外一個(gè)stream的操作，并沒有一個(gè)terminal的操作。如果運(yùn)行上面的代碼的話，則不會(huì)有任何輸出。

? ? 總的來說，對(duì)于一個(gè)stream的操作它會(huì)盡量采用惰性計(jì)算的方式以實(shí)現(xiàn)滿足目標(biāo)結(jié)果。

?

stream執(zhí)行順序

? ? 還有一個(gè)比較值得讓人關(guān)心的就是stream處理元素的執(zhí)行順序。它是按照前面示例里某個(gè)運(yùn)算一次將所有的數(shù)據(jù)處理完之后再傳遞給下一個(gè)呢還是一次處理一個(gè)傳遞下去呢？我們?cè)賮砜慈缦碌拇a：

Stream.of("d2", "a2", "b1", "b3", "c").filter(s -> {System.out.println("filter: " + s);return true;}).forEach(s -> System.out.println("forEach: " + s));

? ? ?運(yùn)行上面這部分程序的輸出如下：

filter: d2 forEach: d2 filter: a2 forEach: a2 filter: b1 forEach: b1 filter: b3 forEach: b3 filter: c forEach: c

? ? 可見，在stream里對(duì)元素的處理是按照流水線的方式來進(jìn)行的。因此它不需要額外的利用集合的數(shù)據(jù)結(jié)構(gòu)來保存中間結(jié)果。這種方式在處理海量數(shù)據(jù)的時(shí)候帶來非常遍歷的特性。

?

Optional類型

? ? Stream api帶來的另外一個(gè)影響就是引入了optional類型的數(shù)據(jù)。關(guān)于optional類型數(shù)據(jù)的詳細(xì)討論會(huì)在后面的文章里描述。這里只是一個(gè)簡單的敘述。我們來看如下的示例：

Optional<Shape> firstBlue = shapes.stream().filter(s -> s.getColor() == BLUE).findFirst();

? ? 在shapes的stream里通過filter方法來過濾一個(gè)符合color == BLUE的元素。實(shí)際上返回的結(jié)果可能存在有這樣的元素，也可能不存在這樣的元素。于是針對(duì)這種可能存在也可能不存在的類型元素，這里引入了Optional類型數(shù)據(jù)來描述它。通過引入Optional類型可以減少和規(guī)避很多容易出現(xiàn)nullpointerexception的情況。也算是對(duì)程序的一種改進(jìn)。

?

stream的潛在并行性

? ? 前面提到過，在stream api里引入了一種使得運(yùn)用并行開發(fā)更加簡便的方式，這就是 parallel stream。在目前多核體系結(jié)構(gòu)比較普遍的情況下，大多數(shù)計(jì)算機(jī)都有多個(gè)核，如果只是使用以前的編程方式的話并不能充分發(fā)揮機(jī)器的性能。于是需要一種更好的方式來使用多核和多線程。在以往的java 多線程開發(fā)里，使用好多線程是一個(gè)很困難的任務(wù)。于是為了簡化對(duì)一些多線程情況下的使用，這里就引入了parallel stram。

? ? 需要注意的是，前面用的stream是對(duì)數(shù)據(jù)進(jìn)行串行處理的，而這里使用并行處理的時(shí)候，它的使用方式則稍微有點(diǎn)差別。我們先來看一部分如下的代碼：

Arrays.asList("a1", "a2", "b1", "c2", "c1").parallelStream().filter(s -> {System.out.format("filter: %s [%s]\n",s, Thread.currentThread().getName());return true;}).map(s -> {System.out.format("map: %s [%s]\n",s, Thread.currentThread().getName());return s.toUpperCase();}).forEach(s -> System.out.format("forEach: %s [%s]\n",s, Thread.currentThread().getName()));

? ? 這部分代碼看起來比較復(fù)雜，實(shí)際上和前面代碼的唯一差別就是stream()方法編程了parallelStream()。在每個(gè)處理步驟里都加入了打印的消息以方便我們跟蹤程序執(zhí)行的過程。如果我們運(yùn)行上述的代碼，會(huì)發(fā)現(xiàn)如下的輸出：

?

filter: b1 [main] map: b1 [main] filter: c2 [ForkJoinPool.commonPool-worker-4] filter: c1 [ForkJoinPool.commonPool-worker-3] map: c1 [ForkJoinPool.commonPool-worker-3] forEach: C1 [ForkJoinPool.commonPool-worker-3] filter: a2 [ForkJoinPool.commonPool-worker-1] map: a2 [ForkJoinPool.commonPool-worker-1] forEach: A2 [ForkJoinPool.commonPool-worker-1] filter: a1 [ForkJoinPool.commonPool-worker-2] map: c2 [ForkJoinPool.commonPool-worker-4] forEach: C2 [ForkJoinPool.commonPool-worker-4] forEach: B1 [main] map: a1 [ForkJoinPool.commonPool-worker-2] forEach: A1 [ForkJoinPool.commonPool-worker-2]

? ? 實(shí)際上，如果我們多次運(yùn)行程序的話會(huì)發(fā)現(xiàn)每次的輸出還有點(diǎn)不一樣。當(dāng)然，從輸出里我們還可以看到一個(gè)東西，就是輸出的線程名是屬于一個(gè)ForkJoinPool里的線程。也就是說它實(shí)際上運(yùn)用了線程池。這里運(yùn)用到的線程池就是java 7里引入的forkjoin pool。關(guān)于forkjoin pool的討論可以參考我前面一篇相關(guān)的分析文章。

?

總結(jié)

? ? Java 8 引入的stream api可以說是給前面函數(shù)式編程應(yīng)用到的lambda表達(dá)式提供了一個(gè)極好的應(yīng)用場(chǎng)景。它本質(zhì)上是一個(gè)惰性計(jì)算流，它不像我們傳統(tǒng)使用的數(shù)據(jù)結(jié)構(gòu)，需要事先分配內(nèi)存空間，而是一種按需計(jì)算的模式。所以它更像是一個(gè)流水線式的計(jì)算模型。同時(shí)，它在默認(rèn)的情況下是串行執(zhí)行的，所以它的執(zhí)行順序不一樣，但是可以利用很少的內(nèi)存空間。另外，在stream api里也有很簡單支持并行計(jì)算的parallemstream，它本質(zhì)上是運(yùn)用了java的Forkjoin thread pool來實(shí)現(xiàn)并行的。這種方式大大簡化了我們并發(fā)編程的難度。

?

參考材料

http://cr.openjdk.java.net/~briangoetz/lambda/lambda-libraries-final.html

http://winterbe.com/posts/2014/07/31/java8-stream-tutorial-examples/

http://www.oracle.com/technetwork/articles/java/ma14-java-se-8-streams-2177646.html

http://www.oracle.com/technetwork/articles/java/architect-streams-pt2-2227132.html

http://www.amazon.com/Java-Action-Lambdas-functional-style-programming/dp/1617291994/ref=sr_1_1?s=books&ie=UTF8&qid=1447684950&sr=1-1&keywords=java+8+in+action

總結(jié)

以上是生活随笔為你收集整理的Java 8 stream学习的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。