1.初識stream
先來一個總綱：
東西就是這么多啦，stream是java8中加入的一個非常實用的功能，最初看時以為是io中的流（其實一點關系都沒有），讓我們先來看一個小例子感受一下：
@Beforepublic void init() { random = new Random(); stuList = new ArrayList<Student>() { { for (int i = 0; i < 100; i++) { add(new Student("student" + i, random.nextInt(50) + 50)); } } };} public class Student { private String name; private Integer score; //-----getters and setters-----} //1列出班上超過85分的學生姓名，并按照分數(shù)降序輸出用戶名字@Testpublic void test1() { List<String> studentList = stuList.stream() .filter(x->x.getScore()>85) .sorted(Comparator.comparing(Student::getScore).reversed()) .map(Student::getName) .collect(Collectors.toList()); System.out.println(studentList);}列出班上分數(shù)超過85分的學生姓名，并按照分數(shù)降序輸出用戶名字，在java8之前我們需要三個步驟：
1）新建一個List<Student> newList，在for循環(huán)中遍歷stuList，將分數(shù)超過85分的學生裝入新的集合中
2）對于新的集合newList進行排序操作
3）遍歷打印newList
這三個步驟在java8中只需要兩條語句，如果緊緊需要打印，不需要保存新生產list的話實際上只需要一條，是不是非常方便。
2.stream的特性
我們首先列出stream的如下三點特性，在之后我們會對照著詳細說明
1.stream不存儲數(shù)據
2.stream不改變源數(shù)據
3.stream的延遲執(zhí)行特性
通常我們在數(shù)組或集合的基礎上創(chuàng)建stream，stream不會專門存儲數(shù)據，對stream的操作也不會影響到創(chuàng)建它的數(shù)組和集合,對于stream的聚合、消費或收集操作只能進行一次，再次操作會報錯，如下代碼：
@Testpublic void test1(){ Stream<String> stream = Stream.generate(()->"user").limit(20); stream.forEach(System.out::println); stream.forEach(System.out::println);}程序在正常完成一次打印工作后報錯。
stream的操作是延遲執(zhí)行的，在列出班上超過85分的學生姓名例子中，在collect方法執(zhí)行之前，filter、sorted、map方法還未執(zhí)行，只有當collect方法執(zhí)行時才會觸發(fā)之前轉換操作
看如下代碼：
public boolean filter(Student s) { System.out.println("begin compare"); return s.getScore() > 85;} @Testpublic void test() { Stream<Student> stream = Stream.of(stuArr).filter(this::filter); System.out.println("split-------------------------------------"); List<Student> studentList = stream.collect(toList());}我們將filter中的邏輯抽象成方法，在方法中加入打印邏輯，如果stream的轉換操作是延遲執(zhí)行的，那么split會先打印，否則后打印，代碼運行結果為
可見stream的操作是延遲執(zhí)行的。
TIP：
當我們操作一個流的時候，并不會修改流底層的集合（即使集合是線程安全的），如果想要修改原有的集合，就無法定義流操作的輸出。
由于stream的延遲執(zhí)行特性，在聚合操作執(zhí)行前修改數(shù)據源是允許的。
List<String> wordList; @Beforepublic void init() { wordList = new ArrayList<String>() { { add("a"); add("b"); add("c"); add("d"); add("e"); add("f"); add("g"); } };}/** * 延遲執(zhí)行特性，在聚合操作之前都可以添加相應元素 */@Testpublic void test() { Stream<String> words = wordList.stream(); wordList.add("END"); long n = words.distinct().count(); System.out.println(n);}最后打印的結果是8
如下代碼是錯誤的
/** * 延遲執(zhí)行特性，會產生干擾 * nullPointException */@Testpublic void test2(){ Stream<String> words1 = wordList.stream(); words1.forEach(s -> { System.out.println("s->"+s); if (s.length() < 4) { System.out.println("select->"+s); wordList.remove(s); System.out.println(wordList); } });}結果報空指針異常
3.創(chuàng)建stream
1）通過數(shù)組創(chuàng)建
/** * 通過數(shù)組創(chuàng)建流 */@Testpublic void testArrayStream(){ //1.通過Arrays.stream //1.1基本類型 int[] arr = new int[]{1,2,34,5}; IntStream intStream = Arrays.stream(arr); //1.2引用類型 Student[] studentArr = new Student[]{new Student("s1",29),new Student("s2",27)}; Stream<Student> studentStream = Arrays.stream(studentArr); //2.通過Stream.of Stream<Integer> stream1 = Stream.of(1,2,34,5,65); //注意生成的是int[]的流 Stream<int[]> stream2 = Stream.of(arr,arr); stream2.forEach(System.out::println);}2）通過集合創(chuàng)建流
/** * 通過集合創(chuàng)建流 */@Testpublic void testCollectionStream(){ List<String> strs = Arrays.asList("11212","dfd","2323","dfhgf"); //創(chuàng)建普通流 Stream<String> stream = strs.stream(); //創(chuàng)建并行流 Stream<String> stream1 = strs.parallelStream();}3）創(chuàng)建空的流
@Testpublic void testEmptyStream(){ //創(chuàng)建一個空的stream Stream<Integer> stream = Stream.empty();}4）創(chuàng)建無限流
@Testpublic void testUnlimitStream(){ //創(chuàng)建無限流，通過limit提取指定大小 Stream.generate(()->"number"+new Random().nextInt()).limit(100).forEach(System.out::println); Stream.generate(()->new Student("name",10)).limit(20).forEach(System.out::println);}5）創(chuàng)建規(guī)律的無限流
/** * 產生規(guī)律的數(shù)據 */@Testpublic void testUnlimitStream1(){ Stream.iterate(0,x->x+1).limit(10).forEach(System.out::println); Stream.iterate(0,x->x).limit(10).forEach(System.out::println); //Stream.iterate(0,x->x).limit(10).forEach(System.out::println);與如下代碼意思是一樣的 Stream.iterate(0, UnaryOperator.identity()).limit(10).forEach(System.out::println);}4.對stream的操作
1）最常使用
map：轉換流，將一種類型的流轉換為另外一種流
/** * map把一種類型的流轉換為另外一種類型的流 * 將String數(shù)組中字母轉換為大寫 */@Testpublic void testMap() { String[] arr = new String[]{"yes", "YES", "no", "NO"}; Arrays.stream(arr).map(x -> x.toLowerCase()).forEach(System.out::println);} filter：過濾流，過濾流中的元素
@Testpublic void testFilter(){ Integer[] arr = new Integer[]{1,2,3,4,5,6,7,8,9,10}; Arrays.stream(arr).filter(x->x>3&&x<8).forEach(System.out::println);} flapMap：拆解流，將流中每一個元素拆解成一個流
/** * flapMap：拆解流 */@Testpublic void testFlapMap1() { String[] arr1 = {"a", "b", "c", "d"}; String[] arr2 = {"e", "f", "c", "d"}; String[] arr3 = {"h", "j", "c", "d"}; // Stream.of(arr1, arr2, arr3).flatMap(x -> Arrays.stream(x)).forEach(System.out::println); Stream.of(arr1, arr2, arr3).flatMap(Arrays::stream).forEach(System.out::println);} sorted：對流進行排序
String[] arr1 = {"abc","a","bc","abcd"};/** * Comparator.comparing是一個鍵提取的功能 * 以下兩個語句表示相同意義 */@Testpublic void testSorted1_(){ /** * 按照字符長度排序 */ Arrays.stream(arr1).sorted((x,y)->{ if (x.length()>y.length()) return 1; else if (x.length()<y.length()) return -1; else return 0; }).forEach(System.out::println); Arrays.stream(arr1).sorted(Comparator.comparing(String::length)).forEach(System.out::println);} /** * 倒序 * reversed(),java8泛型推導的問題，所以如果comparing里面是非方法引用的lambda表達式就沒辦法直接使用reversed() * Comparator.reverseOrder():也是用于翻轉順序，用于比較對象（Stream里面的類型必須是可比較的） * Comparator. naturalOrder()：返回一個自然排序比較器，用于比較對象（Stream里面的類型必須是可比較的） */@Testpublic void testSorted2_(){ Arrays.stream(arr1).sorted(Comparator.comparing(String::length).reversed()).forEach(System.out::println); Arrays.stream(arr1).sorted(Comparator.reverseOrder()).forEach(System.out::println); Arrays.stream(arr1).sorted(Comparator.naturalOrder()).forEach(System.out::println);} /** * thenComparing * 先按照首字母排序 * 之后按照String的長度排序 */@Testpublic void testSorted3_(){ Arrays.stream(arr1).sorted(Comparator.comparing(this::com1).thenComparing(String::length)).forEach(System.out::println);}public char com1(String x){ return x.charAt(0);}2）提取流和組合流
@Before public void init(){ arr1 = new String[]{"a","b","c","d"}; arr2 = new String[]{"d","e","f","g"}; arr3 = new String[]{"i","j","k","l"}; } /** * limit，限制從流中獲得前n個數(shù)據 */ @Test public void testLimit(){ Stream.iterate(1,x->x+2).limit(10).forEach(System.out::println); } /** * skip，跳過前n個數(shù)據 */ @Test public void testSkip(){// Stream.of(arr1).skip(2).limit(2).forEach(System.out::println); Stream.iterate(1,x->x+2).skip(1).limit(5).forEach(System.out::println); } /** * 可以把兩個stream合并成一個stream（合并的stream類型必須相同） * 只能兩兩合并 */ @Test public void testConcat(){ Stream<String> stream1 = Stream.of(arr1); Stream<String> stream2 = Stream.of(arr2); Stream.concat(stream1,stream2).distinct().forEach(System.out::println); }3）聚合操作
@Beforepublic void init(){ arr = new String[]{"b","ab","abc","abcd","abcde"};} /** * max、min * 最大最小值 */@Testpublic void testMaxAndMin(){ Stream.of(arr).max(Comparator.comparing(String::length)).ifPresent(System.out::println); Stream.of(arr).min(Comparator.comparing(String::length)).ifPresent(System.out::println);} /** * count * 計算數(shù)量 */@Testpublic void testCount(){ long count = Stream.of(arr).count(); System.out.println(count);} /** * findFirst * 查找第一個 */@Testpublic void testFindFirst(){ String str = Stream.of(arr).parallel().filter(x->x.length()>3).findFirst().orElse("noghing"); System.out.println(str);} /** * findAny * 找到所有匹配的元素 * 對并行流十分有效 * 只要在任何片段發(fā)現(xiàn)了第一個匹配元素就會結束整個運算 */@Testpublic void testFindAny(){ Optional<String> optional = Stream.of(arr).parallel().filter(x->x.length()>3).findAny(); optional.ifPresent(System.out::println);} /** * anyMatch * 是否含有匹配元素 */@Testpublic void testAnyMatch(){ Boolean aBoolean = Stream.of(arr).anyMatch(x->x.startsWith("a")); System.out.println(aBoolean);} @Testpublic void testStream1() { Optional<Integer> optional = Stream.of(1,2,3).filter(x->x>1).reduce((x,y)->x+y); System.out.println(optional.get());}4）Optional類型
通常聚合操作會返回一個Optional類型，Optional表示一個安全的指定結果類型，所謂的安全指的是避免直接調用返回類型的null值而造成空指針異常，調用optional.ifPresent()可以判斷返回值是否為空，或者直接調用ifPresent(Consumer<? super T> consumer)在結果部位空時進行消費操作；調用optional.get()獲取返回值。通常的使用方式如下：
@Test public void testOptional() { List<String> list = new ArrayList<String>() { { add("user1"); add("user2"); } }; Optional<String> opt = Optional.of("andy with u"); opt.ifPresent(list::add); list.forEach(System.out::println); }使用Optional可以在沒有值時指定一個返回值，例如
@Testpublic void testOptional2() { Integer[] arr = new Integer[]{4,5,6,7,8,9}; Integer result = Stream.of(arr).filter(x->x>9).max(Comparator.naturalOrder()).orElse(-1); System.out.println(result); Integer result1 = Stream.of(arr).filter(x->x>9).max(Comparator.naturalOrder()).orElseGet(()->-1); System.out.println(result1); Integer result2 = Stream.of(arr).filter(x->x>9).max(Comparator.naturalOrder()).orElseThrow(RuntimeException::new); System.out.println(result2);}Optional的創(chuàng)建
采用Optional.empty()創(chuàng)建一個空的Optional，使用Optional.of()創(chuàng)建指定值的Optional。同樣也可以調用Optional對象的map方法進行Optional的轉換，調用flatMap方法進行Optional的迭代
@Testpublic void testStream1() { Optional<Student> studentOptional = Optional.of(new Student("user1",21)); Optional<String> optionalStr = studentOptional.map(Student::getName); System.out.println(optionalStr.get());} public static Optional<Double> inverse(Double x) { return x == 0 ? Optional.empty() : Optional.of(1 / x);} public static Optional<Double> squareRoot(Double x) { return x < 0 ? Optional.empty() : Optional.of(Math.sqrt(x));} /** * Optional的迭代 */@Testpublic void testStream2() { double x = 4d; Optional<Double> result1 = inverse(x).flatMap(StreamTest7::squareRoot); result1.ifPresent(System.out::println); Optional<Double> result2 = Optional.of(4.0).flatMap(StreamTest7::inverse).flatMap(StreamTest7::squareRoot); result2.ifPresent(System.out::println);}5）收集結果
Student[] students;@Beforepublic void init(){ students = new Student[100]; for (int i=0;i<30;i++){ Student student = new Student("user",i); students[i] = student; } for (int i=30;i<60;i++){ Student student = new Student("user"+i,i); students[i] = student; } for (int i=60;i<100;i++){ Student student = new Student("user"+i,i); students[i] = student; } }@Testpublic void testCollect1(){ /** * 生成List */ List<Student> list = Arrays.stream(students).collect(toList()); list.forEach((x)-> System.out.println(x)); /** * 生成Set */ Set<Student> set = Arrays.stream(students).collect(toSet()); set.forEach((x)-> System.out.println(x)); /** * 如果包含相同的key，則需要提供第三個參數(shù)，否則報錯 */ Map<String,Integer> map = Arrays.stream(students).collect(toMap(Student::getName,Student::getScore,(s,a)->s+a)); map.forEach((x,y)-> System.out.println(x+"->"+y));} /** * 生成數(shù)組 */@Testpublic void testCollect2(){ Student[] s = Arrays.stream(students).toArray(Student[]::new); for (int i=0;i<s.length;i++) System.out.println(s[i]);} /** * 指定生成的類型 */@Testpublic void testCollect3(){ HashSet<Student> s = Arrays.stream(students).collect(toCollection(HashSet::new)); s.forEach(System.out::println);} /** * 統(tǒng)計 */@Testpublic void testCollect4(){ IntSummaryStatistics summaryStatistics = Arrays.stream(students).collect(Collectors.summarizingInt(Student::getScore)); System.out.println("getAverage->"+summaryStatistics.getAverage()); System.out.println("getMax->"+summaryStatistics.getMax()); System.out.println("getMin->"+summaryStatistics.getMin()); System.out.println("getCount->"+summaryStatistics.getCount()); System.out.println("getSum->"+summaryStatistics.getSum());}6）分組和分片
分組和分片的意義是，將collect的結果集展示位Map<key,val>的形式，通常的用法如下：
Student[] students;@Beforepublic void init(){ students = new Student[100]; for (int i=0;i<30;i++){ Student student = new Student("user1",i); students[i] = student; } for (int i=30;i<60;i++){ Student student = new Student("user2",i); students[i] = student; } for (int i=60;i<100;i++){ Student student = new Student("user3",i); students[i] = student; } }@Testpublic void testGroupBy1(){ Map<String,List<Student>> map = Arrays.stream(students).collect(groupingBy(Student::getName)); map.forEach((x,y)-> System.out.println(x+"->"+y));} /** * 如果只有兩類，使用partitioningBy會比groupingBy更有效率 */@Testpublic void testPartitioningBy(){ Map<Boolean,List<Student>> map = Arrays.stream(students).collect(partitioningBy(x->x.getScore()>50)); map.forEach((x,y)-> System.out.println(x+"->"+y));} /** * downstream指定類型 */@Testpublic void testGroupBy2(){ Map<String,Set<Student>> map = Arrays.stream(students).collect(groupingBy(Student::getName,toSet())); map.forEach((x,y)-> System.out.println(x+"->"+y));} /** * downstream 聚合操作 */@Testpublic void testGroupBy3(){ /** * counting */ Map<String,Long> map1 = Arrays.stream(students).collect(groupingBy(Student::getName,counting())); map1.forEach((x,y)-> System.out.println(x+"->"+y)); /** * summingInt */ Map<String,Integer> map2 = Arrays.stream(students).collect(groupingBy(Student::getName,summingInt(Student::getScore))); map2.forEach((x,y)-> System.out.println(x+"->"+y)); /** * maxBy */ Map<String,Optional<Student>> map3 = Arrays.stream(students).collect(groupingBy(Student::getName,maxBy(Comparator.comparing(Student::getScore)))); map3.forEach((x,y)-> System.out.println(x+"->"+y)); /** * mapping */ Map<String,Set<Integer>> map4 = Arrays.stream(students).collect(groupingBy(Student::getName,mapping(Student::getScore,toSet()))); map4.forEach((x,y)-> System.out.println(x+"->"+y));} 5.原始類型流
在數(shù)據量比較大的情況下，將基本數(shù)據類型（int,double...）包裝成相應對象流的做法是低效的，因此，我們也可以直接將數(shù)據初始化為原始類型流，在原始類型流上的操作與對象流類似，我們只需要記住兩點
1.原始類型流的初始化
2.原始類型流與流對象的轉換
DoubleStream doubleStream; IntStream intStream; /** * 原始類型流的初始化 */ @Before public void testStream1(){ doubleStream = DoubleStream.of(0.1,0.2,0.3,0.8); intStream = IntStream.of(1,3,5,7,9); IntStream stream1 = IntStream.rangeClosed(0,100); IntStream stream2 = IntStream.range(0,100); } /** * 流與原始類型流的轉換 */ @Test public void testStream2(){ Stream<Double> stream = doubleStream.boxed(); doubleStream = stream.mapToDouble(Double::new); }6.并行流
可以將普通順序執(zhí)行的流轉變?yōu)椴⑿辛?，只需要調用順序流的parallel() 方法即可，如Stream.iterate(1, x -> x + 1).limit(10).parallel()。
1） 并行流的執(zhí)行順序
我們調用peek方法來瞧瞧并行流和串行流的執(zhí)行順序，peek方法顧名思義，就是偷窺流內的數(shù)據，peek方法聲明為Stream<T> peek(Consumer<? super T> action);加入打印程序可以觀察到通過流內數(shù)據，見如下代碼：
public void peek1(int x) { System.out.println(Thread.currentThread().getName() + ":->peek1->" + x); } public void peek2(int x) { System.out.println(Thread.currentThread().getName() + ":->peek2->" + x); } public void peek3(int x) { System.out.println(Thread.currentThread().getName() + ":->final result->" + x); } /** * peek，監(jiān)控方法 * 串行流和并行流的執(zhí)行順序 */ @org.junit.Test public void testPeek() { Stream<Integer> stream = Stream.iterate(1, x -> x + 1).limit(10); stream.peek(this::peek1).filter(x -> x > 5) .peek(this::peek2).filter(x -> x < 8) .peek(this::peek3) .forEach(System.out::println); } @Test public void testPeekPal() { Stream<Integer> stream = Stream.iterate(1, x -> x + 1).limit(10).parallel(); stream.peek(this::peek1).filter(x -> x > 5) .peek(this::peek2).filter(x -> x < 8) .peek(this::peek3) .forEach(System.out::println); }串行流打印結果如下：
并行流打印結果如下：
咋看不一定能看懂，我們用如下的圖來解釋
我們將stream.filter(x -> x > 5).filter(x -> x < 8).forEach(System.out::println)的過程想象成上圖的管道，我們在管道上加入的peek相當于一個閥門，透過這個閥門查看流經的數(shù)據，
1）當我們使用順序流時，數(shù)據按照源數(shù)據的順序依次通過管道，當一個數(shù)據被filter過濾，或者經過整個管道而輸出后，第二個數(shù)據才會開始重復這一過程
2）當我們使用并行流時，系統(tǒng)除了主線程外啟動了七個線程（我的電腦是4核八線程）來執(zhí)行處理任務，因此執(zhí)行是無序的，但同一個線程內處理的數(shù)據是按順序進行的。
2） sorted()、distinct()等對并行流的影響
sorted()、distinct()是元素相關方法，和整體的數(shù)據是有關系的,map，filter等方法和已經通過的元素是不相關的,不需要知道流里面有哪些元素 ，并行執(zhí)行和sorted會不會產生沖突呢？
結論：1.并行流和排序是不沖突的，2.一個流是否是有序的，對于一些api可能會提高執(zhí)行效率，對于另一些api可能會降低執(zhí)行效率
3.如果想要輸出的結果是有序的，對于并行的流需要使用forEachOrdered(forEach的輸出效率更高)
我們做如下實驗：
/** * 生成一億條0-100之間的記錄 */@Beforepublic void init() { Random random = new Random(); list = Stream.generate(() -> random.nextInt(100)).limit(100000000).collect(toList());} /** * tip */@org.junit.Testpublic void test1() { long begin1 = System.currentTimeMillis(); list.stream().filter(x->(x > 10)).filter(x->x<80).count(); long end1 = System.currentTimeMillis(); System.out.println(end1-begin1); list.stream().parallel().filter(x->(x > 10)).filter(x->x<80).count(); long end2 = System.currentTimeMillis(); System.out.println(end2-end1); long begin1_ = System.currentTimeMillis(); list.stream().filter(x->(x > 10)).filter(x->x<80).distinct().sorted().count(); long end1_ = System.currentTimeMillis(); System.out.println(end1-begin1); list.stream().parallel().filter(x->(x > 10)).filter(x->x<80).distinct().sorted().count(); long end2_ = System.currentTimeMillis(); System.out.println(end2_-end1_); }可見，對于串行流.distinct().sorted()方法對于運行時間沒有影響，但是對于串行流，會使得運行時間大大增加，因此對于包含sorted、distinct()等與全局數(shù)據相關的操作，不推薦使用并行流。
7.stream vs spark rdd
最初看到stream的一個直觀感受是和spark像，真的像
val count = sc.parallelize(1 to NUM_SAMPLES).filter { _ => val x = math.random val y = math.random x*x + y*y < 1}.count()println(s"Pi is roughly ${4.0 * count / NUM_SAMPLES}") 以上代碼摘自spark官網，使用的是scala語言，一個最基礎的word count代碼，這里我們簡單介紹一下spark，spark是當今最流行的基于內存的大數(shù)據處理框架，spark中的一個核心概念是RDD（彈性分布式數(shù)據集），將分布于不同處理器上的數(shù)據抽象成rdd，rdd上支持兩種類型的操作1） Transformation（變換）2） Action（行動），對于rdd的Transformation算子并不會立即執(zhí)行，只有當使用了Action算子后，才會觸發(fā)。
關于java8中的stream和spark的相似與不同我們之后專門介紹。
