Spring-Web-Flux實戰(三) - Stream 流

摘要： 0 聯絡我 1. Java開發技術交流Q群 2. 完整部落格連結 3. 個人知乎 4. gayhub 相關原始碼 1 Stream流程式設計-概念 Stream是 Java 8新增加的類，用來補充集合類。 ...

0 聯絡我

1. Java開發技術交流Q群

2. 完整部落格連結

3. 個人知乎

4. gayhub

相關原始碼

1 Stream流程式設計-概念

Stream是 Java 8新增加的類，用來補充集合類。

Stream代表資料流，流中的資料元素的數量可能是有限的，也可能是無限的。

Stream和其它集合類的區別在於

• 其它集合類主要關注與有限數量的資料的訪問和有效管理(增刪改)
• Stream並沒有提供訪問和管理元素的方式，而是通過宣告資料來源的方式，利用可計算的操作在資料來源上執行
  當然BaseStream.iterator() 和 BaseStream.spliterator()操作提供了遍歷元素的方法
• 不儲存資料
  流是基於資料來源的物件，它本身不儲存資料元素，而是通過管道將資料來源的元素傳遞給操作。
  函數語言程式設計。流的操作不會修改資料來源，例如filter不會將資料來源中的資料刪除。
• 延遲操作
  流的很多操作如filter,map等中間操作是延遲執行的，只有到終點操作才會將操作順序執行。
• 可以解綁
  對於無限數量的流，有些操作是可以在有限的時間完成的，比如limit(n) 或 findFirst()，這些操作可是實現"短路"(Short-circuiting)，訪問到有限的元素後就可以返回。
• 純消費
  流的元素只能訪問一次，類似Iterator，操作沒有回頭路，如果你想從頭重新訪問流的元素，對不起，你得重新生成一個新的流

Java Stream提供了提供了序列和並行兩種型別的流，保持一致的介面，提供函數語言程式設計方式，以管道方式提供中間操作和最終執行操作，為Java語言的集合提供了現代語言提供的類似的高階函式操作，簡化和提高了Java集合的功能

2 流的建立

3 流的中間操作

中間操作會返回一個新的流，並且操作是延遲執行的，它不會修改原始資料來源，而是由在終點操作開始的時候才真正開始執行

這和Scala集合的轉換操作不同，Scala集合轉換操作會生成一個新的中間集合，顯而易見Java的這種設計會減少中間物件的生成

3.1 map

將流中的元素對映成另外的值，新的值型別可以和原來的元素的 型別不同

下面的程式碼中將字元元素對映成它的雜湊碼(ASCII值)

List<Integer> l = Stream.of('a','b','c')
.map( c -> c.hashCode())
.collect(Collectors.toList());
System.out.println(l); //[97, 98, 99]

3.2 flatMap

flatmap方法混合了map + flattern的功能，它將對映後的流的元素全部放入到一個新的流中

mapper函式會將每一個元素轉換成一個流物件，而flatMap方法返回的流包含的元素為mapper生成的所有流中的元素

下面這個例子中將一首唐詩生成一個按行分割的流，然後在這個流上呼叫flatmap得到單詞的小寫形式的集合，去掉重複的單詞然後打印出來

String poetry = "Where, before me, are the ages that have gone?\n" +
"And where, behind me, are the coming generations?\n" +
"I think of heaven and earth, without limit, without end,\n" +
"And I am all alone and my tears fall down.";
Stream<String> lines = Arrays.stream(poetry.split("\n"));
Stream<String> words = lines.flatMap(line -> Arrays.stream(line.split(" ")));
List<String> l = words.map( w -> {
if (w.endsWith(",") || w.endsWith(".") || w.endsWith("?"))
return w.substring(0,w.length() -1).trim().toLowerCase();
else
return w.trim().toLowerCase();
}).distinct().sorted().collect(Collectors.toList());,
System.out.println(l); //[ages, all, alone, am, and, are, before, behind, coming, down, earth, end, fall, generations, gone, have, heaven, i, limit, me, my, of, tears, that, the, think, where, without]

3.3 filter

返回的流中只包含滿足斷言(predicate)的資料

下面的程式碼返回流中的偶數集合

List<Integer> l = IntStream.range(1,10)
.filter( i -> i % 2 == 0)
.boxed()
.collect(Collectors.toList());
System.out.println(l); //[2, 4, 6, 8]

3.4 peek

會使用一個Consumer消費流中的元素，但是返回的流還是包含原來的流中的元素。

String[] arr = new String[]{"a","b","c","d"};
Arrays.stream(arr)
.peek(System.out::println) //a,b,c,d
.count();

下面是有狀態操作

3.5 distinct

保證輸出的流中包含唯一的元素，它是通過Object.equals(Object)來檢查是否包含相同的元素

List<String> l = Stream.of("a","b","c","b")
.distinct()
.collect(Collectors.toList());
System.out.println(l); //[a, b, c]

3.6 sorted

將流中的元素按照自然排序方式進行排序，如果元素沒有實現Comparable，則終點操作執行時會丟擲java.lang.ClassCastException異常

sorted(Comparator<? super T> comparator)可以指定排序的方式。

對於有序流，排序是穩定的。對於非有序流，不保證排序穩定。

String[] arr = new String[]{"b_123","c+342","b#632","d_123"};
List<String> l= Arrays.stream(arr)
.sorted((s1,s2) -> {
if (s1.charAt(0) == s2.charAt(0))
return s1.substring(2).compareTo(s2.substring(2));
else
return s1.charAt(0) - s2.charAt(0);
})
.collect(Collectors.toList());
System.out.println(l); //[b_123, b#632, c+342, d_123]

3.7 skip

返回丟棄了前n個元素的流，如果流中的元素小於或者等於n，則返回空的流

3.8 limit

指定數量的元素的流。對於序列流，這個方法是有效的，這是因為它只需返回前n個元素即可，但是對於有序的並行流，它可能花費相對較長的時間，如果你不在意有序，可以將有序並行流轉換為無序的，可以提高效能。

List<Integer> l = IntStream.range(1,100).limit(5)
.boxed()
.collect(Collectors.toList());
System.out.println(l);//[1, 2, 3, 4, 5]

4 流的終止操作

4.1 非短路操作

• forEach、forEachOrdered

  forEach遍歷流的每一個元素，執行指定的action。它是一個終點操作，和peek方法不同。這個方法不擔保按照流的encounter order順序執行，如果對於有序流按照它的encounter order順序執行，你可以使用forEachOrdered方法

Stream.of(1,2,3,4,5).forEach(System.out::println);

• toArray
  將流中的元素放入到一個數組中
• collect
  
  
  
  使用一個collector執行 mutable reduction 操作
  輔助類 Collectors 提供了很多的collector，可以滿足我們日常的需求，你也可以建立新的collector實現特定的需求。它是一個值得關注的類，你需要熟悉這些特定的收集器，如聚合類 averagingInt 、最大最小值 maxBy minBy 、計數 counting 、分組 groupingBy 、字串連線 joining 、分割槽 partitioningBy 、彙總 summarizingInt 、化簡 reducing 、轉換 toXXX 等。

第二個提供了更底層的功能，它的邏輯類似下面的虛擬碼：

R result = supplier.get();
for (T element : this stream)
accumulator.accept(result, element);
return result;

例子

List<String> asList = stringStream.collect(ArrayList::new, ArrayList::add,
ArrayList::addAll);
String concat = stringStream.collect(StringBuilder::new, StringBuilder::append,
StringBuilder::append)
.toString();

• reduce

  是常用的一個方法，事實上很多操作都是基於它實現的。

  它有幾個過載方法

  
  
  

Optional<Integer> total = Stream.of(1,2,3,4,5).reduce( (x, y) -> x +y);
Integer total2 = Stream.of(1,2,3,4,5).reduce(0, (x, y) -> x +y);

值得注意的是accumulator應該滿足結合性(associative)

• max、min

  max返回流中的最大值

  min返回流中的最小值

5 並行流(Parallelism)

所有的流操作都可以序列/並行執行

除非顯示地建立並行流，否則Java庫中建立的都是序列流

Collection.stream() 為集合建立序列流而 Collection.parallelStream() 為集合建立並行流

IntStream.range(int, int) 建立的是序列流

通過parallel()方法可以將序列流轉換成並行流,sequentia()方法將流轉換成序列流。

除非方法的Javadoc中指明瞭方法在並行執行的時候結果是不確定(比如findAny、forEach)，否則序列和並行執行的結果應該是一樣的。

示例