lambda表達式是stream的基礎，初學者建議先學習lambda表達式，http://m.ythuaji.com.cn/article/123637.html
1.初識stream
先來一個總綱：
東西就是這么多啦，stream是java8中加入的一個非常實用的功能，最初看時以為是io中的流（其實一點關系都沒有），讓我們先來看一個小例子感受一下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
@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分的學生姓名，并按照分數降序輸出用戶名字@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);} |
列出班上分數超過85分的學生姓名，并按照分數降序輸出用戶名字，在java8之前我們需要三個步驟：
1）新建一個list<student> newlist，在for循環中遍歷stulist，將分數超過85分的學生裝入新的集合中
2）對于新的集合newlist進行排序操作
3）遍歷打印newlist
這三個步驟在java8中只需要兩條語句，如果緊緊需要打印，不需要保存新生產list的話實際上只需要一條，是不是非常方便。
2.stream的特性
我們首先列出stream的如下三點特性，在之后我們會對照著詳細說明
1.stream不存儲數據
2.stream不改變源數據
3.stream的延遲執行特性
通常我們在數組或集合的基礎上創建stream，stream不會專門存儲數據，對stream的操作也不會影響到創建它的數組和集合,對于stream的聚合、消費或收集操作只能進行一次，再次操作會報錯，如下代碼：
|
1
2
3
4
5
6
|
@testpublic void test1(){ stream<string> stream = stream.generate(()->"user").limit(20); stream.foreach(system.out::println); stream.foreach(system.out::println);} |
程序在正常完成一次打印工作后報錯。
stream的操作是延遲執行的，在列出班上超過85分的學生姓名例子中，在collect方法執行之前，filter、sorted、map方法還未執行，只有當collect方法執行時才會觸發之前轉換操作
看如下代碼：
|
1
2
3
4
5
6
7
8
9
10
11
|
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的轉換操作是延遲執行的，那么split會先打印，否則后打印，代碼運行結果為
可見stream的操作是延遲執行的。
tip：
當我們操作一個流的時候，并不會修改流底層的集合（即使集合是線程安全的），如果想要修改原有的集合，就無法定義流操作的輸出。
由于stream的延遲執行特性，在聚合操作執行前修改數據源是允許的。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
list<string> wordlist; @beforepublic void init() { wordlist = new arraylist<string>() { { add("a"); add("b"); add("c"); add("d"); add("e"); add("f"); add("g"); } };}/** * 延遲執行特性，在聚合操作之前都可以添加相應元素 */@testpublic void test() { stream<string> words = wordlist.stream(); wordlist.add("end"); long n = words.distinct().count(); system.out.println(n);} |
最后打印的結果是8
如下代碼是錯誤的
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
/** * 延遲執行特性，會產生干擾 * 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.創建stream
1）通過數組創建
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
/** * 通過數組創建流 */@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）通過集合創建流
|
1
2
3
4
5
6
7
8
9
10
11
|
/** * 通過集合創建流 */@testpublic void testcollectionstream(){ list<string> strs = arrays.aslist("11212","dfd","2323","dfhgf"); //創建普通流 stream<string> stream = strs.stream(); //創建并行流 stream<string> stream1 = strs.parallelstream();} |
3）創建空的流
|
1
2
3
4
5
6
7
8
9
10
11
12
|
@testpublic void testemptystream(){ //創建一個空的stream stream<integer> stream = stream.empty();}4）創建無限流@testpublic void testunlimitstream(){ //創建無限流，通過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）創建規律的無限流
|
1
2
3
4
5
6
7
8
9
10
|
/** * 產生規律的數據 */@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：轉換流，將一種類型的流轉換為另外一種流
|
1
2
3
4
5
6
7
8
9
|
/** * map把一種類型的流轉換為另外一種類型的流 * 將string數組中字母轉換為大寫 */@testpublic void testmap() { string[] arr = new string[]{"yes", "yes", "no", "no"}; arrays.stream(arr).map(x -> x.tolowercase()).foreach(system.out::println);} |
filter：過濾流，過濾流中的元素
|
1
2
3
4
5
|
@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：拆解流，將流中每一個元素拆解成一個流
|
1
2
3
4
5
6
7
8
9
10
11
|
/** * 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：對流進行排序
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
|
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）提取流和組合流
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
|
@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個數據 */ @test public void testlimit(){ stream.iterate(1,x->x+2).limit(10).foreach(system.out::println); } /** * skip，跳過前n個數據 */ @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）聚合操作
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
|
@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 * 計算數量 */@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 * 找到所有匹配的元素 * 對并行流十分有效 * 只要在任何片段發現了第一個匹配元素就會結束整個運算 */@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()獲取返回值。通常的使用方式如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
|
@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可以在沒有值時指定一個返回值，例如
|
1
2
3
4
5
6
7
8
9
10
|
@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的創建
采用optional.empty()創建一個空的optional，使用optional.of()創建指定值的optional。同樣也可以調用optional對象的map方法進行optional的轉換，調用flatmap方法進行optional的迭代
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
@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）收集結果
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
|
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，則需要提供第三個參數，否則報錯 */ 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));}/** * 生成數組 */@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);}/** * 統計 */@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>的形式，通常的用法如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
|
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.原始類型流
在數據量比較大的情況下，將基本數據類型（int,double...）包裝成相應對象流的做法是低效的，因此，我們也可以直接將數據初始化為原始類型流，在原始類型流上的操作與對象流類似，我們只需要記住兩點
1.原始類型流的初始化
2.原始類型流與流對象的轉換
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
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.并行流
可以將普通順序執行的流轉變為并行流，只需要調用順序流的parallel() 方法即可，如stream.iterate(1, x -> x + 1).limit(10).parallel()。
1） 并行流的執行順序
我們調用peek方法來瞧瞧并行流和串行流的執行順序，peek方法顧名思義，就是偷窺流內的數據，peek方法聲明為stream<t> peek(consumer<? super t> action);加入打印程序可以觀察到通過流內數據，見如下代碼：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
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，監控方法 * 串行流和并行流的執行順序 */ @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相當于一個閥門，透過這個閥門查看流經的數據，
1）當我們使用順序流時，數據按照源數據的順序依次通過管道，當一個數據被filter過濾，或者經過整個管道而輸出后，第二個數據才會開始重復這一過程
2）當我們使用并行流時，系統除了主線程外啟動了七個線程（我的電腦是4核八線程）來執行處理任務，因此執行是無序的，但同一個線程內處理的數據是按順序進行的。
2） sorted()、distinct()等對并行流的影響
sorted()、distinct()是元素相關方法，和整體的數據是有關系的,map，filter等方法和已經通過的元素是不相關的,不需要知道流里面有哪些元素 ，并行執行和sorted會不會產生沖突呢？
結論：1.并行流和排序是不沖突的，2.一個流是否是有序的，對于一些api可能會提高執行效率，對于另一些api可能會降低執行效率
3.如果想要輸出的結果是有序的，對于并行的流需要使用foreachordered(foreach的輸出效率更高)
我們做如下實驗：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
/** * 生成一億條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()等與全局數據相關的操作，不推薦使用并行流。
7.stream vs spark rdd
最初看到stream的一個直觀感受是和spark像，真的像
|
1
2
3
4
|
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是當今最流行的基于內存的大數據處理框架，spark中的一個核心概念是rdd（彈性分布式數據集），將分布于不同處理器上的數據抽象成rdd，rdd上支持兩種類型的操作1） transformation（變換）2） action（行動），對于rdd的transformation算子并不會立即執行，只有當使用了action算子后，才會觸發。
總結
以上所示是小編給大家介紹的java8中的stream相關知識，希望對大家有所幫助，如果大家有任何疑問歡迎給我留言，小編會及時回復大家的！
原文鏈接：http://www.cnblogs.com/andywithu/p/7404101.html
