簡潔的Java8

Stream

標簽 ： Java基礎

---

再次回到阿里, 感覺變化好大: 一是服務資源Docker化, 最牛逼的阿里DB團隊竟然把DB放到了容器中, 還放到了線上環境; 二是全集團Java8(記得離開時還是1.6、1.5, 甚至還有1.4), 在外面創業公司都還停留在1.7的時代, 阿里竟率先使用了Java8, 而且還做了高性能的定制, 因此阿里人也就有機會在生產環境體驗到Java8如絲般的順滑流暢. 而本篇就從對Java8影響最大的Stream開始說起.

---

引入

如果說Runnable接口是將執行邏輯從Thread中剝離了的話, 那Stream則是將數據計算邏輯從Collection中抽離了出來, 使Collection只專注于數據的存儲, 而不用分心計算.

打開Collection Api可以看到多了一個stream() default接口:

/*** Returns a sequential {@code Stream} with this collection as its source.** <p>This method should be overridden when the {@link #spliterator()}* method cannot return a spliterator that is {@code IMMUTABLE},* {@code CONCURRENT}, or <em>late-binding</em>. (See {@link #spliterator()}* for details.)** @implSpec* The default implementation creates a sequential {@code Stream} from the* collection's {@code Spliterator}.** @return a sequential {@code Stream} over the elements in this collection* @since 1.8*/ default Stream<E> stream() {return StreamSupport.stream(spliterator(), false); }

Stream允許以聲明方式處理集合等可以轉換為Stream<T>的數據, 他有很多特點:

• 內部迭代
  與原有的Iterator不同, Stream將迭代操作(類似for/for-each)全部固化到了Api內部實現, 用戶只需傳入表達計算邏輯的lambda表達式(可以理解為Supplier、Function這些的@FunctionalInterface的實現), Stream便會自動迭代數據觸發計算邏輯并生成結果. 內部迭代主要解決了兩方面的問題: 避免集合處理時的套路和晦澀; 便于庫內部實現的多核并行優化.
• 流水線
  很多Stream操作會再返回一個Stream, 這樣多個操作就可以鏈接起來, 形成一個大的流水線, 使其看起來像是對數據源進行數據庫式查詢, 這也就讓自動優化成為可能, 如隱式并行.
• 隱式并行
  如將.stream()替換為.parallelStream(), Stream則會自動啟用Fork/Join框架, 并行執行各條流水線, 并最終自動將結果進行合并.
• 延遲計算
  由于Stream大部分的操作(如filter()、generate()、map()…)都是接受一段lambda表達式, 邏輯類似接口實現(可以看成是回調), 因此代碼并不是立即執行的, 除非流水線上觸發一個終端操作, 否則中間操作不會執行任何處理.
• 短路求值
  有些操作不需要處理整個流就能夠拿到結果, 很多像anyMatch()、allMatch()、limit(), 只要找到一個元素他們的工作就可以結束, 也就沒有必要執行后面的操作, 因此如果后面有大量耗時的操作, 此舉可大大節省性能.

下面一個示例直觀的感受下Stream帶來的便利:

public void joiningList() {// 生成一段[0,20)序列List<Integer> list = IntStream.range(0, 20).boxed().collect(Collectors.toList());// 將list內的偶數提取反向排序后聚合為一個StringString string = list.stream().filter(n -> n % 2 == 0).sorted(Comparator.comparing((Integer i) -> i).reversed()).limit(3).peek((i) -> System.out.println("remained: " + i)).map(String::valueOf).collect(Collectors.joining());System.out.println(string); }

---

Stream 構成

一個流管道(Stream pipeline)通常由3部分構成: 數據源(Source) -> 中間操作/轉換(Transforming) -> 終端操作/執行(Operations): Stream由數據源生成, 經由中間操作串聯起來的一條流水線的轉換, 最后由終端操作觸發執行拿到結果.

Source - 對應Stream的生成: -> 如何生成一個Stream;

Transforming - 對應Stream的轉換: -> 如前面的map()、filter()、limit(), 將原Stream轉換為另一形態;

Operations - 對應Stream的執行: -> 他會真正引發前面一系列Transforming的執行, 并生成一個結果(如List、Array、Optional<T>), 或一個side effect.

我們分別來介紹這些Stream的構成部分:

---

數據源-Stream生成

除了前面介紹過的collection.stream(), 流的生成方式多種多樣, 可簡單概括為3類: 通用流、數值流、其他, 其中以通用流最為常用, 數值流是Java為int、long、double三種數值類型防拆裝箱成本所做的優化:

---

1. 通用流

APIdescription

Arrays.stream(T[] array)	Returns a sequential Stream with the specified array as its source.
Stream.empty()	Returns an empty sequential Stream.
Stream.generate(Supplier<T> s)	Returns an infinite sequential unordered stream where each element is generated by the provided Supplier<T>.
Stream.iterate(T seed, UnaryOperator<T> f)	Returns an infinite sequential ordered Stream produced by iterative application of a function f to an initial element seed, producing a Stream consisting of seed, f(seed), f(f(seed)), etc.
Stream.of(T... values)	Returns a sequential ordered stream whose elements are the specified values.
Stream.concat(Stream<? extends T> a, Stream<? extends T> b)	Creates a lazily concatenated stream whose elements are all the elements of the first stream followed by all the elements of the second stream.
StreamSupport.stream(Spliterator<T> spliterator, boolean parallel)	Creates a new sequential or parallel Stream from a Spliterator.

---

2. 數值流

APIdescription

Arrays.stream(Xxx[] array)	Returns a sequential Int/Long/DoubleStream with the specified array as its source.
XxxStream.empty()	Returns an empty sequential Int/Long/DoubleStream.
XxxStream.generate(XxxSupplier s)	Returns an infinite sequential unordered stream where each element is generated by the provided Int/Long/DoubleSupplier.
XxxStream.iterate(Xxx seed, XxxUnaryOperator f)	Returns an infinite sequential ordered Int/Long/DoubleStream like as Stream.iterate(T seed, UnaryOperator<T> f)
XxxStream.of(Xxx... values)	Returns a sequential ordered stream whose elements are the specified values.
XxxStream.concat(XxxStream a, XxxStream b)	Creates a lazily concatenated stream whose elements are all the elements of the first stream followed by all the elements of the second stream.
Int/LongStream.range(startInclusive, endExclusive)	Returns a sequential ordered Int/LongStream from startInclusive (inclusive) to endExclusive (exclusive) by an incremental step of 1.
Int/LongStream.rangeClosed(startInclusive, endInclusive)	Returns a sequential ordered Int/LongStream from startInclusive (inclusive) to endInclusive (inclusive) by an incremental step of 1.

---

3. 其他

• I/O Stream
  
  • BufferedReader.lines()
• File Stream
  
  • Files.lines(Path path)
  • Files.find(Path start, int maxDepth, BiPredicate<Path,BasicFileAttributes> matcher, FileVisitOption... options)
  • DirectoryStream<Path> newDirectoryStream(Path dir)
  • Files.walk(Path start, FileVisitOption... options)
• Jar
  
  • JarFile.stream()
• Random
  
  • Random.ints()
  • Random.longs()
  • Random.doubles()
• Pattern
  
  • splitAsStream(CharSequence input)
    …

另外, 三種數值流之間, 以及數值流與通用流之間都可以相互轉換:
1. 數值流轉換: doubleStream.mapToInt(DoubleToIntFunction mapper)、intStream.asLongStream() …
2. 數值流轉通用流: longStream.boxed()、intStream.mapToObj(IntFunction<? extends U> mapper) …
3. 通用流轉數值流: stream.flatMapToInt(Function<? super T,? extends IntStream> mapper)、stream.mapToDouble(ToDoubleFunction<? super T> mapper)…

---

中間操作-Stream轉換

所有的中間操作都會返回另一個Stream, 這讓多個操作可以鏈接起來組成中間操作鏈, 從而形成一條流水線, 因此它的特點就是前面提到的延遲執行: 觸發流水線上觸發一個終端操作, 否則中間操作不執行任何處理.

APIDescription

filter(Predicate<? super T> predicate)	Returns a stream consisting of the elements of this stream that match the given predicate.
distinct()	Returns a stream consisting of the distinct elements (according to Object.equals(Object)) of this stream.
limit(long maxSize)	Returns a stream consisting of the elements of this stream, truncated to be no longer than maxSize in length.
skip(long n)	Returns a stream consisting of the remaining elements of this stream after discarding the first n elements of the stream.
sorted(Comparator<? super T> comparator)	Returns a stream consisting of the elements of this stream, sorted according to the provided Comparator.
map(Function<? super T,? extends R> mapper)	Returns a stream consisting of the results of applying the given function to the elements of this stream.
flatMap(Function<? super T,? extends Stream<? extends R>> mapper)	Returns a stream consisting of the results of replacing each element of this stream with the contents of a mapped stream produced by applying the provided mapping function to each element.
peek(Consumer<? super T> action)	Returns a stream consisting of the elements of this stream, additionally performing the provided action on each element as elements are consumed from the resulting stream.

這里著重講解下flatMap(), 因為我在第一次接觸他時也沒明白他到底能做什么:
假設我們有這樣一個字符串list:List<String> strs = Arrays.asList("hello", "alibaba", "world");如何列出里面各不相同的字符呢?
首先我們想到的是String包含一個split()方法, 將字符串分解為子串, 于是我們這樣寫:

Stream<Stream<String>> streamStream = strs.stream().map(str -> Arrays.stream(str.split("")));

我們將String分解成String[]后再由Arrays.stream()將String[]映射成Stream<String>, 但這個結果是我們不想看到的: 我們明明想要的是Stream<String>卻得到的是Stream<Stream<String>>, 他把我們想要的結果包到Stream里面了. 這時候就需要我們的flatMap()出場了:

Stream<String> stringStream = strs.stream().flatMap(str -> Arrays.stream(str.split("")));

flatMap()把Stream中的層級結構扁平化了, 將內層Stream內的元素抽取出來, 最終新的Stream就沒有內層Stream了.

可以簡單概括為: flatMap()方法讓你把一個流中的每個值都換成另一個Stream, 然后把所有的Stream連接起來成為一個Stream.

---

終端操作-Stream執行

終端操作不僅擔負著觸發流水線執行的任務, 他還需要拿到流水線執行的結果, 其結果為任何不是流的值, 如List、Array、boolean、Optional<T>, 甚至是void(forEach()):

ApiDescription

count()	Returns the count of elements in this stream.
max(Comparator<? super T> comparator)	Returns the maximum element of this stream according to the provided Comparator.
min(Comparator<? super T> comparator)	Returns the minimum element of this stream according to the provided Comparator.
allMatch(Predicate<? super T> predicate)	Returns whether all elements of this stream match the provided predicate.
anyMatch(Predicate<? super T> predicate)	Returns whether any elements of this stream match the provided predicate.
noneMatch(Predicate<? super T> predicate)	Returns whether no elements of this stream match the provided predicate.
findAny()	Returns an Optional describing some element of the stream, or an empty Optional if the stream is empty.
findFirst()	Returns an Optional describing the first element of this stream, or an empty Optional if the stream is empty.
reduce(BinaryOperator<T> accumulator)	Performs a reduction on the elements of this stream, using an associative accumulation function, and returns an Optional describing the reduced value, if any.
toArray()	Returns an array containing the elements of this stream.
forEach(Consumer<? super T> action)	Performs an action for each element of this stream.
forEachOrdered(Consumer<? super T> action)	Performs an action for each element of this stream, in the encounter order of the stream if the stream has a defined encounter order.
collect(Collector<? super T,A,R> collector)	Performs a mutable reduction operation on the elements of this stream using a Collector.

像IntStream/LongStream/DoubleStream還提供了average()、sum()、summaryStatistics()這樣的操作, 拿到一個對Stream進行匯總了的結果.

---

other

java.util.stream.Stream接口集成自java.util.stream.BaseStream接口, 而BaseStream接口也提供了很多工具方法(如將串行流轉換為并行流的parallel()方法)供我們使用:

ApiDescription

S onClose(Runnable closeHandler)	Returns an equivalent stream with an additional close handler.
void close()	Closes this stream, causing all close handlers for this stream pipeline to be called.
S unordered()	Returns an equivalent stream that is unordered.
Iterator<T> iterator()	Returns an iterator for the elements of this stream.
Spliterator<T> spliterator()	Returns a spliterator for the elements of this stream.
S sequential()	Returns an equivalent stream that is sequential.
S parallel()	Returns an equivalent stream that is parallel.
boolean isParallel()	Returns whether this stream, if a terminal operation were to be executed, would execute in parallel.

---

綜合實戰

下面, 我們針對一系列交易提出一些問題綜合實踐上面列舉的Api:

• DO定義

/*** 交易員*/ private class Trader {private String name;private String city;public Trader(String name, String city) {this.name = name;this.city = city;}public String getName() {return name;}public String getCity() {return city;}@Overridepublic String toString() {return "Trader{" +"name='" + name + '\'' +", city='" + city + '\'' +'}';} }/*** 交易*/ private class Transaction {private Trader trader;private int year;private int value;public Transaction(Trader trader, int year, int value) {this.trader = trader;this.year = year;this.value = value;}public Trader getTrader() {return this.trader;}public int getYear() {return this.year;}public int getValue() {return this.value;}@Overridepublic String toString() {return "Transaction{" +"trader=" + trader +", year=" + year +", value=" + value +'}';} }

• Stream操作

/*** @author jifang.zjf* @since 2017/7/3 下午4:05.*/ public class StreamLambda {private List<Transaction> transactions;@Beforepublic void setUp() {Trader raoul = new Trader("Raoul", "Cambridge");Trader mario = new Trader("Mario", "Milan");Trader alan = new Trader("Alan", "Cambridge");Trader brian = new Trader("Brian", "Cambridge");transactions = Arrays.asList(new Transaction(brian, 2011, 300),new Transaction(raoul, 2012, 1000),new Transaction(raoul, 2011, 400),new Transaction(mario, 2012, 710),new Transaction(mario, 2012, 700),new Transaction(alan, 2012, 950));}@Testpublic void action() {// 1. 打印2011年發生的所有交易, 并按交易額排序(從低到高)transactions.stream().filter(transaction -> transaction.getYear() == 2011).sorted(Comparator.comparing(Transaction::getValue)).forEachOrdered(System.out::println);// 2. 找出交易員都在哪些不同的城市工作過Set<String> distinctCities = transactions.stream().map(transaction -> transaction.getTrader().getCity()).collect(Collectors.toSet()); // or .distinct().collect(Collectors.toList())System.out.println(distinctCities);// 3. 找出所有來自于劍橋的交易員, 并按姓名排序Trader[] traders = transactions.stream().map(Transaction::getTrader).filter(trader -> trader.getCity().equals("Cambridge")).distinct().sorted(Comparator.comparing(Trader::getName)).toArray(Trader[]::new);System.out.println(Arrays.toString(traders));// 4. 返回所有交易員的姓名字符串, 并按字母順序排序String names = transactions.stream().map(transaction -> transaction.getTrader().getName()).distinct().sorted(Comparator.naturalOrder()).reduce("", (str1, str2) -> str1 + " " + str2);System.out.println(names);// 5. 返回所有交易員的姓名字母串, 并按字母順序排序String letters = transactions.stream().map(transaction -> transaction.getTrader().getName()).distinct().map(name -> name.split("")).flatMap(Arrays::stream).sorted().collect(Collectors.joining());System.out.println(letters);// 6. 有沒有交易員是在米蘭工作boolean workMilan = transactions.stream().anyMatch(transaction -> transaction.getTrader().getCity().equals("Milan"));System.out.println(workMilan);// 7. 打印生活在劍橋的交易員的所有交易額總和long sum = transactions.stream().filter(transaction -> transaction.getTrader().getCity().equals("Cambridge")).mapToLong(Transaction::getValue).sum();System.out.println(sum);// 8. 所有交易中，最高的交易額是多少OptionalInt max = transactions.stream().mapToInt(Transaction::getValue).max();// or transactions.stream().map(Transaction::getValue).max(Comparator.naturalOrder());System.out.println(max.orElse(0));// 9. 找到交易額最小的交易Optional<Transaction> min = transactions.stream().min(Comparator.comparingInt(Transaction::getValue));System.out.println(min.orElseThrow(IllegalArgumentException::new));} } 參考

Java 8新特性：全新的Stream API

Java 8 中的 Streams API 詳解

Java 8：不要再用循環了

Java 8 in Action

---

• by 菜鳥-翡青
  
  • 博客: 菜鳥-翡青 - http://blog.csdn.net/zjf280441589
  • 微博: 菜鳥-翡青 - http://weibo.com/u/3319050953
  • 另: 阿里巴巴-菜鳥網絡長期招人, 有意向的同學可直接email我的郵箱: jifang.zjf@alibaba-inc.com, 備注請注明意向BU, 謝謝.

總結

以上是生活随笔為你收集整理的简洁的Java8的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。