[[430750]]

面試官：java8新增的stream api用過嗎?

我：這個必須用過啊。

面試官：給你下面一個字符串數組,如果用stream api來實現，找出以字符'a'開頭長度最大的字符串，使用stream api該怎么實現呢?

{"abb","abcd","fegc","efe","adfes"} 

我：用下面這個方法來實現：

public static void maxLength(List<String> list){ 
    System.out.println(list.stream().filter(s -> s.startsWith("a")).mapToInt(r -> length(r)).max().orElse(0));; 
} 

面試官：這個操作是迭代一次還是迭代兩次呢?也就是說是先迭代一遍，過濾出以字符'a'開頭的字符串數組，然后再迭代一次，找出最大長度，還是一次迭代完成呢?

我:這個是迭代一次完成，如果要是迭代多次，stream后面的操作函數很多的情況下效率會非常低。我們加個打印可以來驗證結果，代碼如下：

public static void main(String[] args) { 
    List<String> list = Arrays.asList("abb", "abcd", "fegc", "efe", "adfes"); 
    int maxLength = list.stream(). 
          filter(s -> isStartWitha(s)). 
    mapToInt(StreamTest1::length). 
    max().orElse(0); 
    System.out.println("以字符a開頭的字符串最大長度:" + maxLength); 
} 
 
private static boolean isStartWitha(String a){ 
    System.out.println(a + " is start with a：" + a.startsWith("a")); 
    return a.startsWith("a"); 
} 
 
private static int length(String a){ 
    System.out.println("the length of" + a + ":" + a.length()); 
    return a.length(); 
} 

打印結果如下：

abb is start with a：true 
the length of abb:3 
abcd is start with a：true 
the length of abcd:4 
fegc is start with a：false 
efe is start with a：false 
adfes is start with a：true 
the length of adfes:5 
以字符a開頭的字符串最大長度:5 

面試官：你確定只是迭代一次嗎?有其他情況嗎?

我：有。filter是一個無狀態的中間操作，對于這個中間操作來說，stream處理只需要迭代一次。但是對于有狀態的中間操作，就需要迭代多次。

面試官：你剛剛提到有狀態的操作和無狀態的操作，這個是怎么區分呢?

我：在stream api中，無狀態的操作是指當前元素的操作不受前面元素的影響，主要包括如下方法：

filter(),flatMap(),flatMapToInt(),flatMapToLong(),flatMapToDouble(),map(),mapToInt(),mapToDouble(),mapToLong(),peek(),unordered() 

而有狀態的操作是指需要等所有元素處理完之后才能執行當前操作，主要包括下面方法：

distinct(),limit(),skip(),sorted(),sorted()  

面試官：有狀態的操作，能舉個例子嗎?

我：比如下面這段代碼：

public static void main(String[] args) { 
    List<Integer> list = Arrays.asList(5, 2, 3, 1, 4); 
 
    List<Integer> newArray = list.stream() 
            .map(StreamTest2::map1) 
            .sorted((o1, o2) -> o1 - o2) 
            .map(StreamTest2::map2) 
            .collect(Collectors.toList()); 
    System.out.println("新的有序數組:" + newArray); 
} 
 
private static Integer map1(Integer i) { 
    int result = i * 10; 
    System.out.println("線程:" + Thread.currentThread().getName() + " 方法map1入參：" + i + ",輸出:" + result); 
    return result; 
} 
 
private static Integer map2(Integer i) { 
    int result = i * 10; 
    System.out.println("線程:" + Thread.currentThread().getName() + " 方法map2入參：" + i + ",輸出:" + result); 
    return result; 
} 

上面代碼中，對原始數組進行了兩次迭代，第一次迭代對所有數組元素都調用了map1方法乘以10，然后對新數組進行排序，第二次迭代對排序后的數組元素調用map2方法，即對排序后的數組元素乘以10。方法輸出如下：

線程:main 方法map1入參：5,輸出:50 
線程:main 方法map1入參：2,輸出:20 
線程:main 方法map1入參：3,輸出:30 
線程:main 方法map1入參：1,輸出:10 
線程:main 方法map1入參：4,輸出:40 
線程:main 方法map2入參：10,輸出:100 
線程:main 方法map2入參：20,輸出:200 
線程:main 方法map2入參：30,輸出:300 
線程:main 方法map2入參：40,輸出:400 
線程:main 方法map2入參：50,輸出:500 
新的有序數組:[100, 200, 300, 400, 500] 

面試官：了解過底層原理嗎?

我：我來先畫一下Stream的UML類圖：

這個類圖說明以下幾點：

• AbstractPipeline有基本類型的子類，如LongPipeline和DoublePipeline，還有一個引用類型的子類ReferencePipeline。
• 無論是ReferencePipeline，還是LongPipeline和DoublePipeline等基本類型的Pipeline，都有3個內部類來繼承自己。
• StatelessOp對應無狀態的操作，StatefulOp對應有狀態的操作，Head對應Collection.stream()方法返回結果。
• 無論是StatelessOp、StatefulOp還是Head，都是一個Pipeline，這些Pipeline用雙向鏈表串聯起來，每個Pipeline節點被看作一個Stage，Head是鏈表的頭結點。上面UML類圖中AbstractPipeline類中previousStage和nextStage就代表雙向鏈表當前節點指向前后節點的引用。如下圖：

面試官：上面用雙向鏈表把所有操作都串聯起來了，這樣可以實現從Head節點開始依次執行所有的操作。但是這些操作怎么疊加在一起呢?比如下面這段代碼有三個map方法，后面的方法要依賴前面的計算結果：

List<Integer> list = Arrays.asList(5, 2, 3, 1, 4); 
List<Integer> newArray = list.stream().map(StreamTest2::map1).map(StreamTest2::map2).map(StreamTest2::map3).collect(Collectors.toList()); 

我：Stream提供了Sink接口來處理操作的疊加。上面代碼的map方法把操作封裝到了Sink，每個節點執行操作時，調用Sink的accept方法就可以把操作結果傳給下一個節點的Sink。比如map方法源代碼如下：

public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) { 
    Objects.requireNonNull(mapper); 
    return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE, 
                                 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) { 
        @Override 
    //返回包裝成的Sink 
        Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) { 
            return new Sink.ChainedReference<P_OUT, R>(sink) { 
                @Override 
                public void accept(P_OUT u) { 
            //downstream是下游節點的Sink，把當前節點的執行結果傳給下游節點 
                    downstream.accept(mapper.apply(u)); 
                } 
            }; 
        } 
    }; 
} 

面試官：能詳細講一下Sink嗎?

我：Sink主要提供了下面4個方法

//執行操作之前調用這個方法 
void begin(long size) 
//執行操作之后調用這個方法 
void end() 
//是否可以結束操作 
boolean cancellationRequested() 
//操作執行函數 
void accept() 

對于有狀態的操作，必須實現begin和end兩個方法，因為begin方法會創建一個存放中間結果的容器，accept方法將元素放入該容器，end方法負責對容器中元素處理，比如排序。

面試官：那cancellationRequested方法什么時候用呢?

我：這個方法用于短路操作，比如stream.findAny。

面試官：你剛剛提到短路操作，怎么區分短路操作和非短路操作呢?

我：短路操作和非短路操作都是Stream的結束操作，結束操作是針對中間操作來說的。短路操作是指不用處理全部元素就可以結束，包括下面的方法：

anyMatch(),allMatch(),noneMatch(),findFirst(),findAny() 

非短路操作是指需要處理所有元素才能結束，包括下面的方法：

forEach(),forEachOrdered(),toArray(),reduce(),collect(),max(),min(),count() 

總結一下Stream操作，如下圖：

在遇到結束操作時，所有Pipeline節點封裝的Sink會串成一個鏈表，如下圖：

把Sink串成鏈表的過程可以參考下面這段源代碼：

final <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) { 
    Objects.requireNonNull(sink); 
    for ( @SuppressWarnings("rawtypes") AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) { 
        sink = p.opWrapSink(p.previousStage.combinedFlags, sink); 
    } 
    return (Sink<P_IN>) sink; 
} 

這樣從Head節點開始依次調用每個節點封裝的Sink中的begin,accept,cancellationRequested,end 四個方法就可以完成Steam流水線的執行。

面試官：上面提到了Sink會串成一個鏈，那對于有返回結果的操作，返回的結果是保存在什么地方呢?

我：這里分三種情況：

• 如果返回結果是boolean(比如 anyMatch、allMatch、noneMatch)和Optional(比如 findFirst、findAny)，返回結果存放在對應的Sink。
• collect, reduce等規約操作，返回結果存放在用戶指定的容器中，比如如下代碼返回結果放在Optional容器中：

Optional accResult = Stream.of(1, 2, 3, 4, 5).reduce((sum, item) -> { 
    sum += item; 
    return sum; 
}); 

max 和 min也是規約操作，因為底層是通過調用 reduce 方法實現的。

• 對于返回是數組的情況，返回數組之前，數據會存放在一種多叉樹數據結構中，這種多叉樹結構元素存儲在樹的葉子當中，一個葉子節點可以存放多個元素。

面試官：上面你提到返回數組的時候用到了多叉樹的結構，這樣做對于Stream處理有什么好處呢?

我：按照官方的說法，這樣做是為了避免在并行操作期間不必要地復制數據。

面試官：能簡單介紹一下Stream的并行處理嗎?

我：Stream的并行處理用到了Fork/Join框架，如下圖：

計算過程中，先把任務拆解成子任務，并行計算。計算完成后再把子任務計算結果合并成結果集。

面試官：Fork/Join框架跟普通線程池相比，有什么優勢嗎?

我：fork/join框架的優勢是, 如果某個子任務需要等待另外一個子任務完成才能繼續工作，那么處理線程會主動尋找其他未完成的子任務進行執行。跟普通線程池相比，減少了等待時間。

面試官：使用Stream并行流，一定會比串行快嗎?

我：這個不一定，使用的時候要考慮以下幾個因素：

• 要處理的元素數量，數據越多，性能提升越明顯。
• 數據結構的可分割性，數組、ArrayList支持隨機讀取，可分割性好，HashSet、TreeSet雖然可以分割，但不太容易分割均勻，LinkedList、Streams.iterate、BufferedReader.lines因為長度未知，可分解性差。
• 盡量使用基本類型，避免裝箱拆箱。
• 單個子任務花費時間越長，帶來的性能提升就會越大。

面試官：據說Stream api跟普通迭代相比有性能損耗，你怎么看?

我：對于簡單的處理操作，Stream api性能確實不如普通迭代。但是如果CPU性能好的話，使用Stream并行處理性能會明細提高。對于復雜處理操作，無論并行還是串行，Stream api有明顯的優勢。

對于并行處理，要考慮CPU的核數。