摘要：運行機制分為源，中間操作，終止操作。反過來說，目前還無法專為某個并行流指定這個值。我們在本節(jié)中已經(jīng)指出，并行流不總是比順序流快。特別是和等依賴于元素順序的操作，它們在并行流上執(zhí)行的代價非常大。

Stream是一組用來處理數(shù)組，集合的API。

不是數(shù)據(jù)結(jié)構(gòu)，沒有內(nèi)部存儲。

不支持索引訪問。

延遲計算

支持并行

很容易生成數(shù)據(jù)或集合

支持過濾，查找，轉(zhuǎn)換，匯總，聚合等操作。

Stream分為源source，中間操作，終止操作。

流的源可以是一個數(shù)組，集合，生成器方法，I/O通道等等。

一個流可以有零個或多個中間操作，每一個中間操作都會返回一個新的流，供下一個操作使用，一個流只會有一個終止操作。

Stream只有遇到終止操作，它的源才會開始執(zhí)行遍歷操作。

通過數(shù)組,Stream.of()

通過集合

通過Stream.generate方法來創(chuàng)建

通過Stram.iterate方法

其他API

    import java.util.Arrays;
    import java.util.List;
    import java.util.stream.IntStream;
    import java.util.stream.Stream;
    
    public class CreateStream {
    
    	//通過數(shù)組,Stream.of()
        static void gen1(){
            String[] str = {"a","b","c"};
            Stream str1 = Stream.of(str);
        }
    
    	//通過集合
        static void gen2(){
            List strings = Arrays.asList("a", "b", "c");
            Stream stream = strings.stream();
        }
    
    	//通過Stream.generate方法來創(chuàng)建
        static void gen3(){
        	//這是一個無限流，通過這種方法創(chuàng)建在操作的時候最好加上limit進(jìn)行限制
            Stream generate = Stream.generate(() -> 1);
            generate.limit(10).forEach(x -> System.out.println(x));
        }
        
    	//通過Stram.iterate方法
        static void gen4(){
            Stream iterate = Stream.iterate(1, x -> x +1);
            iterate.forEach(x -> System.out.println(x));
        }
    
    	//其他API
        static void gen5(){
            String str = "abc";
            IntStream chars = str.chars();
            chars.forEach(x -> System.out.println(x));
        }
    }

2.1.1 filter 過濾

該操作會接受一個謂詞（一個返回boolean的函數(shù)）作為參數(shù)，并返回一個包括所有符合謂詞的元素的流。說白了就是給一個條件，filter會根據(jù)這個條件截取流中得數(shù)據(jù)。

public static void testFilter(){
        List integers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        //截取所有能被2整除得數(shù)據(jù)
        List collect = integers.stream().filter(i -> i % 2 == 0).collect(Collectors.toList());
        System.out.println("collect = " + collect);
    }

結(jié)果：

collect = [2, 4, 6, 8, 10]

2.1.2 distinct 去重

該操作會返回一個元素各異（根據(jù)流所生成元素的hashCode和equals方法實現(xiàn)）的流。

public static void main(String[] args) {
        List numbers = Arrays.asList(1, 2, 1, 3, 3, 2, 4);
        List collect = numbers.stream().distinct().collect(Collectors.toList());
        System.out.println("collect = " + collect);
}

結(jié)果：

collect = [1, 2, 3, 4]

2.1.3 sorted 排序

對流中得數(shù)據(jù)進(jìn)行排序，可以以自然序或著用Comparator 接口定義的排序規(guī)則來排序一個流。Comparator 能使用lambada表達(dá)式來初始化，還能夠逆序一個已經(jīng)排序的流。

    public static void main(String[] args) {
        List integers = Arrays.asList(5, 8, 2, 6, 41, 11);
        //排序默認(rèn)為順序  順序 = [2, 5, 6, 8, 11, 41]
        List sorted = integers.stream().sorted().collect(Collectors.toList());
        System.out.println("順序 = " + sorted);
        //逆序    逆序 = [41, 11, 8, 6, 5, 2]
        List reverseOrder = integers.stream().sorted(Comparator.reverseOrder()).collect(Collectors.toList());
        System.out.println("逆序 = " + reverseOrder);
        //也可以接收一個lambda
        List ages = integers.stream().sorted(Comparator.comparing(User::getAge)).collect(Collectors.toList());
    }

2.1.4 limit 截取

該方法會返回一個不超過給定長度的流。

public static void testLimit(){
        List integers = Arrays.asList(1, 2, 1, 3, 3, 2, 4);
        //截取流中得前三個元素  collect = [1, 2, 1]
        List collect = integers.stream().limit(3).collect(Collectors.toList());
        System.out.println("collect = " + collect);
    }

2.1.5 skip 舍棄

該方法會返回一個扔掉了前面n個元素的流。如果流中元素不足n個，則返回一個空流。

public static void testSkip(){
        List integers = Arrays.asList(1, 2, 1, 3, 3, 2, 4);
        //丟掉流中得前三個元素  collect = [3, 3, 2, 4]
        List collect = integers.stream().skip(3).collect(Collectors.toList());
        System.out.println("collect = " + collect);
}

2.1.6 map 歸納

該方法會接受一個函數(shù)作為參數(shù)，這個函數(shù)會被應(yīng)用到每個元素上，并將其映射成一個新的元素。就是根據(jù)指定函數(shù)獲取流中得每個元素得數(shù)據(jù)并重新組合成一個新的元素。

public static void main(String[] args) {
        //自己建好得一個獲取對象list得方法
        List dishList = Dish.getDishList();
        //獲取每一道菜得名稱  并放到一個list中
        List collect = dishList.stream().map(Dish::getName).collect(Collectors.toList());
        //collect = [pork, beef, chicken, french fries, rice, season fruit, pizza, prawns, salmon]
        System.out.println("collect = " + collect);
    }

2.1.7 flatMap 扁平化

該方法key可以讓你把一個流中的每個值都換成另一個流，然后把所有的流都鏈接起來成為一個流。

給 定 單 詞 列 表["Hello","World"] ，你想要返回列表 ["H","e","l", "o","W","r","d"],你可能會認(rèn)為這很容易，通過map你可以把每個單詞映射成一張字符表，然后調(diào)用 distinct 來過濾重復(fù)的字符，但是這個方法的問題在于，傳遞給 map 方法的Lambda為每個單詞返回了一個 String[] （ String列表）。因此， map 返回的流實際上是Stream 類型的。而你真正想要的是用Stream 來表示一個字符流。

正確寫法應(yīng)該是通過flatMap對其扁平化并作出對應(yīng)處理。

public static void main(String[] args) {
        String[] words = {"Hello", "World"};
        List collect = Stream.of(words).        //數(shù)組轉(zhuǎn)換流
                map(w -> w.split("")).  //去掉“”并獲取到兩個String[]
                flatMap(Arrays::stream).        //方法調(diào)用將兩個String[]扁平化為一個stream
                distinct().                     //去重    
                collect(Collectors.toList());
        //collect = [H, e, l, o, W, r, d]
        System.out.println("collect = " + collect);
    }
}

2.1.8 peek

peek 的設(shè)計初衷就是在流的每個元素恢復(fù)運行之前，插入執(zhí)行一個動作。

public static void main(String[] args) {
        List numbers = Arrays.asList(2, 3, 4, 5);
        List result =
                numbers.stream()
                        .peek(x -> System.out.println("from stream: " + x))
                        .map(x -> x + 17)
                        .peek(x -> System.out.println("after map: " + x))
                        .filter(x -> x % 2 == 0)
                        .peek(x -> System.out.println("after filter: " + x))
                        .limit(3)
                        .peek(x -> System.out.println("after limit: " + x))
                        .collect(Collectors.toList());
        
    }

結(jié)果：

from stream: 2
after map: 19
from stream: 3
after map: 20
after filter: 20
after limit: 20
from stream: 4
after map: 21
from stream: 5
after map: 22
after filter: 22
after limit: 22

2.1.9 collect 收集

從上面得代碼已經(jīng)可以看出來，collect是將最終stream中得數(shù)據(jù)收集起來，最終生成一個list，set，或者map。

public static void main(String[] args) {
        List dishList = Dish.getDishList();
        //list
        List collect = dishList.stream().limit(2).collect(Collectors.toList());
        //set
        Set collect1 = dishList.stream().limit(2).collect(Collectors.toSet());
        //map
        Map collect2 = dishList.stream().limit(2).collect(Collectors.toMap(Dish::getName, Dish::getType));
}

這里面生成map得toMap方法有三個重載，傳入得參數(shù)都不同，這里使用得是傳入兩個Function類型得參數(shù)。當(dāng)然，Collectors的功能還不止這些，下面的收集器中會有其他的詳解。

循環(huán) forEach

計算 min、max、count、average

匹配 anyMatch、allMatch、noneMatch、findFirst、findAny

匯聚 reduce

收集器 collect

另一個常見的數(shù)據(jù)處理套路是看看數(shù)據(jù)集中的某些元素是否匹配一個給定的屬性。Stream API通過allMatch，anyMatch，noneMatch，findFirst和findAny方法提供了這樣的工具。

查找和匹配都是終端操作。

2.3.1 anyMatch

anyMatch方法可以回答“流中是否有一個元素能匹配到給定的謂詞”。會返回一個boolean值。

public class AnyMatch {
    public static void main(String[] args) {
        List dish = Dish.getDish();
        boolean b = dish.stream().anyMatch(Dish::isVegetarian);
        System.out.println(b);
    }
}

2.3.2 allMatch

allMatch方法和anyMatch類似，校驗流中是否都能匹配到給定的謂詞。

class AllMatch{
    public static void main(String[] args) {
        List dish = Dish.getDish();
        //是否所有菜的熱量都小于1000
        boolean b = dish.stream().allMatch(d -> d.getCalories() < 1000);
        System.out.println(b);
    }
}

2.3.3 noneMatch

noneMatch方法可以確保流中沒有任何元素與給定的謂詞匹配。

class NoneMatch{
    public static void main(String[] args) {
        List dish = Dish.getDish();
        //沒有任何菜的熱量大于等于1000
        boolean b = dish.stream().allMatch(d -> d.getCalories() >= 1000);
        System.out.println(b);
    }
}

anyMatch，noneMatch，allMatch這三個操作都用到了所謂的短路。

2.3.4 findAny

findAny方法將返回當(dāng)前流中的任意元素。

class FindAny{
    public static void main(String[] args) {
        List dish = Dish.getDish();
        Optional any = dish.stream().filter(Dish::isVegetarian).findAny();
        System.out.println("any = " + any);
    }
}

2.3.5 findFirst

findFirst方法能找到你想要的第一個元素。

class FindFirst{
        public static void main(String[] args) {
            List dish = Dish.getDish();
            Optional any = dish.stream().filter(Dish::isVegetarian).findFirst();
            System.out.println("any = " + any);
        }
    }

此類查詢需要將流中所有元素反復(fù)結(jié)合起來，得到一個值，比如一個 Integer 。這樣的查詢可以被歸類為歸約操作（將流歸約成一個值）。用函數(shù)式編程語言的術(shù)語來說，這稱為折疊（fold），因為你可以將這個操 作看成把一張長長的紙（你的流）反復(fù)折疊成一個小方塊，而這就是折疊操作的結(jié)果。

2.4.1 元素求和

public static void main(String[] args) {
        List integers = Arrays.asList(1, 2, 3, 6, 8);
        //求list中的和，以0為基數(shù)
        Integer reduce = integers.stream().reduce(0, (a, b) -> a + b);
        //Integer的靜態(tài)方法
        int sum = integers.stream().reduce(0, Integer::sum);
        System.out.println("reduce = " + reduce);
    }

2.4.2 最大值和最小值

public static void main(String[] args) {
        List integers = Arrays.asList(1, 2, 3, 6, 8);
        Optional min = integers.stream().reduce(Integer::min);
        System.out.println("min = " + min);
        Optional max = integers.stream().reduce(Integer::max);
        System.out.println("max = " + max);
    }

2.5.1 查找流中的最大值和最小值 minBy maxBy

public static void main(String[] args) {
        List dish = Dish.getDish();
        //創(chuàng)建一個Comparator來進(jìn)行比較  比較菜的卡路里
        Comparator dishComparator = Comparator.comparingInt(Dish::getCalories);
        //maxBy選出最大值
        Optional collect = dish.stream().collect(Collectors.maxBy(dishComparator));
        System.out.println("collect = " + collect);
        //選出最小值
        Optional collect1 = dish.stream().collect(Collectors.minBy(dishComparator));
        System.out.println("collect1 = " + collect1);
    }

2.5.2 匯總 summingInt

Collectors.summingInt 。它可接受一個把對象映射為求和所需 int 的函數(shù)，并返回一個收集器。

 public static void main(String[] args) {
        List dish = Dish.getDish();
        //計算總和
        int collect = dish.stream().collect(Collectors.summingInt(Dish::getCalories));
        System.out.println("collect = " + collect);
    }

2.5.3 平均數(shù) averagingInt

public static void main(String[] args) {
        List dish = Dish.getDish();
        //計算平均數(shù)
        Double collect = dish.stream().collect(Collectors.averagingInt(Dish::getCalories));
        System.out.println("collect = " + collect);
    }

2.5.4 連接字符串 joining

public static void main(String[] args) {
        List dish = Dish.getDish();
        String collect = dish.stream().map(Dish::getName).collect(Collectors.joining());
        System.out.println("collect = " + collect);
    }

joining 工廠方法有一個重載版本可以接受元素之間的分界符，這樣你就可以得到一個逗號分隔的菜肴名稱列表。

String collect = dish.stream().map(Dish::getName).collect(Collectors.joining(", "));

2.5.5 得到流中的總數(shù) counting

long howManyDishes = dish.stream().collect(Collectors.counting());

2.6.1 分組 groupingBy

public static void main(String[] args) {
        List dish = Dish.getDish();
        //groupingBy接受一個function作為參數(shù)
        Map> collect = dish.stream().collect(Collectors.groupingBy(Dish::getType));
        System.out.println("collect = " + collect);
    }

如果想用以分類的條件可能比簡單的屬性訪問器要復(fù)雜。例如，你可能想把熱量不到400卡路里的菜劃分為“低熱量”（diet），熱量400到700卡路里的菜劃為“普通”（normal），高于700卡路里的劃為“高熱量”（fat）。由于 Dish 類的作者沒有把這個操作寫成一個方法，你無法使用方法引用，但你可以把這個邏輯寫成Lambda表達(dá)式。

public static void main(String[] args) {
        List dishList = Dish.getDish();
        Map> collect = dishList.stream().collect(Collectors.groupingBy(dish->{
            if (dish.getCalories() <= 400) {
                return "DIET";
            } else if (dish.getCalories() <= 700) {
                return "NORMAL";
            } else {
                return "FAT";
            }
        }));
        System.out.println("collect = " + collect);
    }

2.6.2 多級分組

要實現(xiàn)多級分組，我們可以使用一個由雙參數(shù)版本的 Collectors.groupingBy 工廠方法創(chuàng)建的收集器，它除了普通的分類函數(shù)之外，還可以接受 collector 類型的第二個參數(shù)。那么要進(jìn)行二級分組的話，我們可以把一個內(nèi)層groupingBy 傳遞給外層 groupingBy ，并定義一個為流中項目分類的二級標(biāo)準(zhǔn)。

public static void main(String[] args) {
        List dish = Dish.getDish();
        Map>> collect = dish.stream().collect(Collectors.groupingBy(Dish::getType, Collectors.groupingBy(d -> {
            if (d.getCalories() <= 400) {
                return "DIET";
            } else if (d.getCalories() <= 700) {
                return "NORMAL";
            } else {
                return "FAT";
            }
        })));
        System.out.println("collect = " + collect);
    }

2.6.3 按子組收集數(shù)據(jù)

在上一面，我們看到可以把第二個 groupingBy 收集器傳遞給外層收集器來實現(xiàn)多級分組。但進(jìn)一步說，傳遞給第一個 groupingBy 的第二個收集器可以是任何類型，而不一定是另一個 groupingBy 。

例如，要數(shù)一數(shù)菜單中每類菜有多少個，可以傳遞 counting 收集器作為groupingBy 收集器的第二個參數(shù)。

Map typesCount = dish.stream().collect(groupingBy(Dish::getType, counting()));

普通的單參數(shù) groupingBy(f) （其中 f 是分類函數(shù)）實際上是 groupingBy(f,toList()) 的簡便寫法。

把收集器的結(jié)果轉(zhuǎn)換為另一種類型:

Collectors.collectingAndThen工廠方法

并行流就是一個把內(nèi)容分成多個數(shù)據(jù)塊，并用不同的線程分別處理每個數(shù)據(jù)塊的流。這樣一來，你就可以自動把給定操作的工作負(fù)荷分配給多核處理器的所有內(nèi)核，讓它們都忙起來。

你可以把流轉(zhuǎn)換成并行流，從而讓前面的函數(shù)歸約過程（也就是求和）并行運行——對順序流調(diào)用 parallel 方法：

public static long parallelSum(long n) {
        return Stream.iterate(1L, i -> i + 1)
                .limit(n)
                .parallel()
                .reduce(0L, Long::sum);
    }

Stream 在內(nèi)部分成了幾塊。因此可以對不同的塊獨立并行進(jìn)行歸納操作，最后，同一個歸納操作會將各個子流的部分歸納結(jié)果合并起來，得到整個原始流的歸納結(jié)果。

配置并行流使用的線程池

看看流的 parallel 方法，你可能會想，并行流用的線程是從哪兒來的？有多少個？怎么自定義這個過程呢？

并行流內(nèi)部使用了默認(rèn)的 ForkJoinPool （7.2節(jié)會進(jìn)一步講到分支/合并框架），它默認(rèn)的線 程 數(shù) 量 就是 你 的 處 理器 數(shù) 量 ， 這個 值 是 由 Runtime.getRuntime().available-Processors() 得到的。 但 是 你 可 以 通 過 系 統(tǒng) 屬 性 java.util.concurrent.ForkJoinPool.common.parallelism 來改變線程池大小，如下所示： System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism","12"); 這是一個全局設(shè)置，因此它將影響代碼中所有的并行流。反過來說，目前還無法專為某個并行流指定這個值。一般而言，讓 ForkJoinPool 的大小等于處理器數(shù)量是個不錯的默認(rèn)值，除非你有很好的理由，否則我們強烈建議你不要修改它。

我們聲稱并行求和方法應(yīng)該比順序和迭代方法性能好。然而在軟件工程上，靠猜絕對不是什么好辦法！特別是在優(yōu)化性能時，你應(yīng)該始終遵循三個黃金規(guī)則：測量，測量，再測量。

如果有疑問，測量。把順序流轉(zhuǎn)成并行流輕而易舉，但卻不一定是好事。我們在本節(jié)中已經(jīng)指出，并行流不總是比順序流快。此外，并行流有時候會和你的直覺不一致，所以在考慮選擇順序流還是并行流時，第一個也是最重要的建議就是用適當(dāng)?shù)幕鶞?zhǔn)來檢查其性能。

留意裝箱。自動裝箱和拆箱操作會大大降低性能。Java 8中有原始類型流（ IntStream 、LongStream 、 DoubleStream ）來避免這種操作，但凡有可能都應(yīng)該用這些流。

有些操作本身在并行流上的性能就比順序流差。特別是 limit 和 findFirst 等依賴于元素順序的操作，它們在并行流上執(zhí)行的代價非常大。例如， findAny 會比 findFirst 性能好，因為它不一定要按順序來執(zhí)行。你總是可以調(diào)用 unordered 方法來把有序流變成無序流。那么，如果你需要流中的n個元素而不是專門要前n個的話，對無序并行流調(diào)用limit 可能會比單個有序流（比如數(shù)據(jù)源是一個 List ）更高效。

還要考慮流的操作流水線的總計算成本。設(shè)N是要處理的元素的總數(shù)，Q是一個元素通過流水線的大致處理成本，則N*Q就是這個對成本的一個粗略的定性估計。Q值較高就意味著使用并行流時性能好的可能性比較大。

對于較小的數(shù)據(jù)量，選擇并行流幾乎從來都不是一個好的決定。并行處理少數(shù)幾個元素的好處還抵不上并行化造成的額外開銷。

要考慮流背后的數(shù)據(jù)結(jié)構(gòu)是否易于分解。例如， ArrayList 的拆分效率比 LinkedList高得多，因為前者用不著遍歷就可以平均拆分，而后者則必須遍歷。另外，用 range 工廠方法創(chuàng)建的原始類型流也可以快速分解。

流自身的特點，以及流水線中的中間操作修改流的方式，都可能會改變分解過程的性能。例如，一個 SIZED 流可以分成大小相等的兩部分，這樣每個部分都可以比較高效地并行處理，但篩選操作可能丟棄的元素個數(shù)卻無法預(yù)測，導(dǎo)致流本身的大小未知。

還要考慮終端操作中合并步驟的代價是大是小（例如 Collector 中的 combiner 方法）。如果這一步代價很大，那么組合每個子流產(chǎn)生的部分結(jié)果所付出的代價就可能會超出通過并行流得到的性能提升。

3.3 分支/合并框架

分支/合并框架的目的是以遞歸方式將可以并行的任務(wù)拆分成更小的任務(wù)，然后將每個子任務(wù)的結(jié)果合并起來生成整體結(jié)果。它是 ExecutorService 接口的一個實現(xiàn)，它把子任務(wù)分配給線程池（稱為 ForkJoinPool ）中的工作線程。

3.3.1 使用RecursiveTask

要把任務(wù)提交到這個池，必須創(chuàng)建 RecursiveTask 的一個子類，其中 R 是并行化任務(wù)（以 及所有子任務(wù)）產(chǎn)生的結(jié)果類型，或者如果任務(wù)不返回結(jié)果，則是 RecursiveAction 類型（當(dāng) 然它可能會更新其他非局部機構(gòu)）。

要定義 RecursiveTask， 只需實現(xiàn)它唯一的抽象方法compute ：

protected abstract R compute();

這個方法同時定義了將任務(wù)拆分成子任務(wù)的邏輯，以及無法再拆分或不方便再拆分時，生成單個子任務(wù)結(jié)果的邏輯。

3.3.2 使用RecursiveTask求和

public class ForkJoinSumCalculator
        extends java.util.concurrent.RecursiveTask {
    private final long[] numbers;
    private final int start;
    private final int end;
    public static final long THRESHOLD = 10_000;

    public ForkJoinSumCalculator(long[] numbers) {
        this(numbers, 0, numbers.length);
    }

    private ForkJoinSumCalculator(long[] numbers, int start, int end) {
        this.numbers = numbers;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        int length = end - start;
        if (length <= THRESHOLD) {
            return computeSequentially();
        }
        //創(chuàng)建一個子任務(wù)來為數(shù)組得前一半求和
        ForkJoinSumCalculator leftTask =
                new ForkJoinSumCalculator(numbers, start, start + length / 2);
        //利 用 另 一 個ForkJoinPool線程異步執(zhí)行新創(chuàng)建的子任務(wù)
        leftTask.fork();
        //創(chuàng)建一個子任務(wù)來為數(shù)組得后一半求和
        ForkJoinSumCalculator rightTask =
                new ForkJoinSumCalculator(numbers, start + length / 2, end);
        //同步執(zhí)行第二個子任務(wù)，有可能進(jìn)一步遞歸
        Long rightResult = rightTask.compute();
        //讀取第一個任務(wù)得結(jié)構(gòu)，未完成就等待
        Long leftResult = leftTask.join();
        return leftResult + rightResult;
    }

    private long computeSequentially() {
        long sum = 0;
        for (int i = start; i < end; i++) {
            sum += numbers[i];
        }
        return sum;
    }

    public static long forkJoinSum(long n) {
        long[] numbers = LongStream.rangeClosed(1, n).toArray();
        ForkJoinTask task = new ForkJoinSumCalculator(numbers);
        return new ForkJoinPool().invoke(task);
    }

    public static void main(String[] args) {
        long l = ForkJoinSumCalculator.forkJoinSum(5);
        System.out.println("l = " + l);
    }

}