摘要：前言系列神秘的系列神奇的函數(shù)式接口繼上兩篇之后，本文已經(jīng)系列的第三篇了。相反，他們會返回一個持有結(jié)果的新。操作是延遲執(zhí)行的。截斷流，使其元素不超過給定數(shù)量。返回流中元素總數(shù)。返回流中最大值。

「Java8系列」神秘的Lambda

「Java8系列」神奇的函數(shù)式接口

繼上兩篇之后，本文已經(jīng)java8系列的第三篇了。本篇文章比較長，但我希望大家都能認真讀完。讀不完可以先收藏，在找時間讀。沒看過前兩篇的可以點上邊的鏈接看看，前兩篇文章算是對是用Stream鋪墊的一點基礎(chǔ)吧，不過不看也可以學(xué)會使用Stream，但看了會有助于更好的理解和使用。在沒有深入了解之前，我以為Stream也是數(shù)據(jù)的載體，但后來發(fā)現(xiàn)并不是。那么它到底是什么？聽我慢慢道來。

Stream它并不是一個容器，它只是對容器的功能進行了增強，添加了很多便利的操作,例如查找、過濾、分組、排序等一系列的操作。并且有串行、并行兩種執(zhí)行模式，并行模式充分的利用了多核處理器的優(yōu)勢，使用fork/join框架進行了任務(wù)拆分，同時提高了執(zhí)行速度。簡而言之，Stream就是提供了一種高效且易于使用的處理數(shù)據(jù)的方式。

特點：

Stream自己不會存儲元素。

Stream的操作不會改變源對象。相反，他們會返回一個持有結(jié)果的新Stream。

Stream 操作是延遲執(zhí)行的。它會等到需要結(jié)果的時候才執(zhí)行。也就是執(zhí)行終端操作的時候。

圖解：

一個Stream的操作就如上圖，在一個管道內(nèi)，分為三個步驟，第一步是創(chuàng)建Stream，從集合、數(shù)組中獲取一個流，第二步是中間操作鏈，對數(shù)據(jù)進行處理。第三步是終端操作，用來執(zhí)行中間操作鏈，返回結(jié)果。

由集合創(chuàng)建：

Java8 中的 Collection 接口被擴展，提供了兩個獲取流的方法,這兩個方法是default方法，也就是說所有實現(xiàn)Collection接口的接口都不需要實現(xiàn)就可以直接使用：

default Stream stream() : 返回一個順序流。

default Stream parallelStream() : 返回一個并行流。

例如：
    List integerList = new ArrayList<>();
    integerList.add(1);
    integerList.add(2);
    Stream stream = integerList.stream();
    Stream stream1 = integerList.parallelStream();

由數(shù)組創(chuàng)建：

Java8 中的 Arrays 的靜態(tài)方法 stream() 可以獲取數(shù)組流:

static Stream stream(T[] array): 返回一個流

重載形式，能夠處理對應(yīng)基本類型的數(shù)組:

public static IntStream stream(int[] array)
public static LongStream stream(long[] array)
public static DoubleStream stream(double[] array)

例如：
    int[] intArray = {1,2,3};
    IntStream stream = Arrays.stream(intArray);

由值創(chuàng)建:

可以使用靜態(tài)方法 Stream.of(), 通過顯示值 創(chuàng)建一個流。它可以接收任意數(shù)量的參數(shù)。

public static Stream of(T... values) : 返回一個流。

例如：
Stream integerStream = Stream.of(1, 2, 3, 4, 5, 6, 7, 8);

由函數(shù)創(chuàng)建：創(chuàng)建無限流

可以使用靜態(tài)方法 Stream.iterate() 和 Stream.generate()創(chuàng)建無限流。

迭代
public static Stream iterate(final T seed, final UnaryOperator f)

生成
public static Stream generate(Supplier s)

 例如：
 Stream.generate(Math::random).limit(5).forEach(System.out::print); 
 List collect = Stream.iterate(0,i -> i + 1).limit(5).collect(Collectors.toList());

注意：使用無限流一定要配合limit截斷，不然會無限制創(chuàng)建下去。

如果Stream只有中間操作是不會執(zhí)行的，當執(zhí)行終端操作的時候才會執(zhí)行中間操作，這種方式稱為延遲加載或惰性求值。多個中間操作組成一個中間操作鏈，只有當執(zhí)行終端操作的時候才會執(zhí)行一遍中間操作鏈，具體是因為什么我們在后面再說明。下面看下Stream有哪些中間操作。

Stream distinct()：

去重，通過流所生成元素的 hashCode() 和 equals() 去除重復(fù)元素。

Stream filter(Predicate predicate):

Predicate函數(shù)在上一篇當中我們已經(jīng)講過，它是斷言型接口，所以filter方法中是接收一個和Predicate函數(shù)對應(yīng)Lambda表達式，返回一個布爾值，從流中過濾某些元素。

Stream sorted(Comparator comparator):

指定比較規(guī)則進行排序。

Stream limit(long maxSize):

截斷流，使其元素不超過給定數(shù)量。如果元素的個數(shù)小于maxSize，那就獲取所有元素。

Stream skip(long n):

跳過元素，返回一個扔掉了前 n 個元素的流。若流中元素不足 n 個，則返回一個空流。與 limit(n) 互補。

Stream map(Function mapper)：

接收一個Function函數(shù)作為參數(shù)，該函數(shù)會被應(yīng)用到每個元素上，并將其映射成一個新的元素。也就是轉(zhuǎn)換操作，map還有三個應(yīng)用于具體類型方法，分別是：mapToInt，mapToLong和mapToDouble。這三個方法也比較好理解，比如mapToInt就是把原始Stream轉(zhuǎn)換成一個新的Stream，這個新生成的Stream中的元素都是int類型。這三個方法可以免除自動裝箱/拆箱的額外消耗。

Stream flatMap(Function> mapper)：

接收一個Function函數(shù)作為參數(shù)，將流中的每個值都轉(zhuǎn)換成另一個流，然后把所有流連接成一個流。flatMap也有三個應(yīng)用于具體類型的方法，分別是：flatMapToInt、flatMapToLong、flatMapToDouble，其作用于map的三個衍生方法相同。

終端操作執(zhí)行中間操作鏈，并返回結(jié)果。終端操作我們就不一一介紹了，只介紹一下常用的操作。詳細可看java.util.stream.Stream接口中的方法。

void forEach(Consumer action):

內(nèi)部迭代(需要用戶去做迭代，稱為外部迭代。相反，Stream API使用內(nèi)部迭代幫你把迭代做了)

 users.stream().forEach(user -> System.out.println(user.getName()));

R collect(Collector collector):

收集、將流轉(zhuǎn)換為其他形式，比如轉(zhuǎn)換成List、Set、Map。collect方法是用Collector作為參數(shù)，Collector接口中方法的實現(xiàn)決定了如何對流執(zhí)行收集操作(如收集到 List、Set、Map)。但是 Collectors 實用類提供了很多靜態(tài)方法，可以方便地創(chuàng)建常見收集器實例。例舉一些常用的：

 List users = Lists.newArrayList();
 users.add(new User(15, "A", ImmutableList.of("1元", "5元")));
 users.add(new User(25, "B", ImmutableList.of("10元", "50元")));
 users.add(new User(21, "C", ImmutableList.of("100元")));
 //收集名稱到List
 List nameList = users.stream().map(User::getName).collect(Collectors.toList());
 //收集名稱到List
 Set nameSet = users.stream().map(User::getName).collect(Collectors.toSet());
 //收集到map,名字作為key,user對象作為value
 Map userMap = users.stream()
                .collect(Collectors.toMap(User::getName, Function.identity(), (k1, k2) -> k2));

其他終端操作：

boolean allMatch(Predicate predicate); 檢查是否匹配所有元素。

boolean anyMatch(Predicate predicate); 檢查是否至少匹配一個元素。

boolean noneMatch(Predicate predicate); 檢查是否沒有匹配所有元素。

Optional findFirst(); 返回當前流中的第一個元素。

Optional findAny(); 返回當前流中的任意元素。

long count(); 返回流中元素總數(shù)。

Optional max(Comparator comparator); 返回流中最大值。

Optional min(Comparator comparator); 返回流中最小值。

T reduce(T identity, BinaryOperator accumulator); 可以將流中元素反復(fù)結(jié)合起來，得到一個值。 返回 T。這是一個歸約操作。

上面我們提到過，說Stream的并行模式使用了Fork/Join框架，這里簡單說下Fork/Join框架是什么？Fork/Join框架是java7中加入的一個并行任務(wù)框架，可以將任務(wù)拆分為多個小任務(wù)，每個小任務(wù)執(zhí)行完的結(jié)果在合并成為一個結(jié)果。在任務(wù)的執(zhí)行過程中使用工作竊取（work-stealing）算法，減少線程之間的競爭。

Fork/Join圖解

工作竊取圖解

先看下整體類圖:藍色箭頭代表繼承，綠色箭頭代表實現(xiàn)，紅色箭頭代表內(nèi)部類。

實際上Stream只有兩種操作，中間操作、終端操作，中間操作只是一種標記，只有終端操作才會實際觸發(fā)執(zhí)行。所以Stream流水線式的操作大致應(yīng)該是用某種方式記錄中間操作，只有調(diào)用終端操作才會將所有的中間操作疊加在一起在一次迭代中全部執(zhí)行。這里只做簡單的介紹，想詳細了解的可以參考下面的參考資料中的鏈接。

操作怎么記錄？

Stream的操作記錄是通過ReferencePipeline記錄的，ReferencePipeline有三個內(nèi)部類Head、StatelessOp、StatefulOp，Stream中使用Stage的概念來描述一個完整的操作，并用某種實例化后的ReferencePipeline來代表Stage，Head用于表示第一個Stage，即調(diào)用諸如Collection.stream()方法產(chǎn)生的Stage，很顯然這個Stage里不包含任何操作，StatelessOp和StatefulOp分別表示無狀態(tài)和有狀態(tài)的Stage，對應(yīng)于無狀態(tài)和有狀態(tài)的中間操作。

操作怎么疊加？

操作是記錄完了，但是前面的Stage并不知道后面Stage到底執(zhí)行了哪種操作，以及回調(diào)函數(shù)是哪種形式。這就需要有某種協(xié)議來協(xié)調(diào)相鄰Stage之間的調(diào)用關(guān)系。
這種協(xié)議由Sink接口完成，Sink接口包含的方法如下表所示:

void begin(long size)，開始遍歷元素之前調(diào)用該方法，通知Sink做好準備。

void end()，所有元素遍歷完成之后調(diào)用，通知Sink沒有更多的元素了。

boolean cancellationRequested()，是否可以結(jié)束操作，可以讓短路操作盡早結(jié)束。

void accept(T t)，遍歷元素時調(diào)用，接受一個待處理元素，并對元素進行處理。Stage把自己包含的操作和回調(diào)方法封裝到該方法里，前一個Stage只需要調(diào)用當前Stage.accept(T t)方法就行了。

每個Stage都會將自己的操作封裝到一個Sink里，前一個Stage只需調(diào)用后一個Stage的accept()方法即可，并不需要知道其內(nèi)部是如何處理的。有了Sink對操作的包裝，Stage之間的調(diào)用問題就解決了，執(zhí)行時只需要從流水線的head開始對數(shù)據(jù)源依次調(diào)用每個Stage對應(yīng)的Sink.{begin(), accept(), cancellationRequested(), end()}方法就可以了。

操作怎么執(zhí)行？

Sink完美封裝了Stream每一步操作，并給出了[處理->轉(zhuǎn)發(fā)]的模式來疊加操作。這一連串的齒輪已經(jīng)咬合，就差最后一步撥動齒輪啟動執(zhí)行。是什么啟動這一連串的操作呢？也許你已經(jīng)想到了啟動的原始動力就是結(jié)束操作(Terminal Operation)，一旦調(diào)用某個結(jié)束操作，就會觸發(fā)整個流水線的執(zhí)行。

https://ifeve.com/stream

https://www.ibm.com/developer...
https://segmentfault.com/a/11...
https://github.com/CarpenterL...