Lambda表達(dá)式

• JVM內(nèi)部是通過(guò)invokedynamic指令來(lái)實(shí)現(xiàn)Lambda表達(dá)式的
• Lambda中允許將一個(gè)函數(shù)作為方法的參數(shù),即函數(shù)作為參數(shù)傳遞進(jìn)方法中
• 使用Lambda表達(dá)式可以使代碼更加簡(jiǎn)潔

變量作用域

• Lambda表達(dá)式只能引用標(biāo)記了final的外層局部變量.即不能在Lambda表達(dá)式內(nèi)部修改定義在作用域外的局部變量,否則會(huì)導(dǎo)致報(bào)錯(cuò)
• Lambda表達(dá)式中可以直接訪問(wèn)外層的局部變量
• Lambda表達(dá)式中外層局部變量可以不用聲明為final, 但是必須不可被后面的代碼修改,即隱性地具有final的語(yǔ)義
• Lambda表達(dá)式中不允許聲明一個(gè)與外層局部變量同名的參數(shù)或者局部變量

使用示例

匿名內(nèi)部類(lèi)

• 匿名內(nèi)部類(lèi): 匿名內(nèi)部類(lèi)仍然是一個(gè)類(lèi),不需要指定類(lèi)名,編譯器會(huì)自動(dòng)為該類(lèi)取名
  • Java中的匿名內(nèi)部類(lèi):

  public class MainAnonymousClass {
    public static void main(String[] args) {
        new Thread(new Runnable(){
            @Override
            public void run(){
                System.out.println("Anonymous Class Thread run()");
            }
        }).start();;
    }
  }
  

  • 使用Lambda表達(dá)式實(shí)現(xiàn)匿名內(nèi)部類(lèi):

  public class MainLambda {
    public static void main(String[] args) {
        new Thread(
                () -> System.out.println("Lambda Thread run()")
            ).start();;
    }
  }
  

帶參函數(shù)

• 帶參函數(shù)的簡(jiǎn)寫(xiě):

List<String> list = Arrays.asList("I", "love", "you", "too");
Collections.sort(list, new Comparator<String>() { // 接口名
    @Override
    public int compare(String s1, String s2) { // 方法名
        if(s1 == null)
            return -1;
        if(s2 == null)
            return 1;
        return s1.length() - s2.length();           
    }
});

• 上述代碼通過(guò)內(nèi)部類(lèi)重載了Comparator接口的compare() 方法來(lái)實(shí)現(xiàn)比較邏輯. 采用Lambda表達(dá)式可簡(jiǎn)寫(xiě)如下:

List<String> list = Arrays.asList("I", "love", "you", "too");
Collections.sort(list, (s1, s2) -> { // 省略參數(shù)表類(lèi)型
    if (s1 == null)
        return -1;
    if (s2 == null)
        return 1;
    return s1.length() - s2.length();
});

• 上述代碼根內(nèi)部類(lèi)的作用一樣
• 除了省略了接口名和方法名,代碼中的參數(shù)類(lèi)型也可以省略
• 因?yàn)?strong>javac的類(lèi)型推斷機(jī)制,編譯器能夠根據(jù)上下文信息推斷出參數(shù)的類(lèi)型

Collection

forEach

• 增強(qiáng)型for循環(huán):

ArrayList<String> list = new ArrayList<>(Arrays.asList("I", "love", "you", "too"));
for(String str : list) {
    if (str.length() > 3)
        System.out.println(str);
}

• 使用forEach() 方法結(jié)合匿名內(nèi)部類(lèi)實(shí)現(xiàn):

ArrayList<String> list = new ArrayList<>(Arrays.asList("I", "love", "you", "too"));
list.forEach(new Consumer<String>(){
    @Override
    public void accept(String str) {
        if (str.length() > 3) {
            System.out.println(str);
        }
    }
});

• 使用Lambda表達(dá)式實(shí)現(xiàn)如下:

// 使用forEach()結(jié)合Lambda表達(dá)式迭代
ArrayList<String> list = new ArrayList<>(Arrays.asList("I", "love", "you", "too"));
list.forEach(str -> {
    if (str.length() > 3) {
        Systemm.out.println(str);
    }       
});

上述代碼給forEach() 方法傳入一個(gè)Lambda表達(dá)式,不需要知道accept() 方法,也不需要知道Consumer接口,類(lèi)型推導(dǎo)已經(jīng)完成了這些

removeIf

• 該方法簽名: boolean removeIf(Predicate<? super E> filter);
  • 刪除容器中所有滿足filter指定條件的元素
    • Predicate是一個(gè)函數(shù)接口,里面有一個(gè)待實(shí)現(xiàn)的方法boolean test(T t)
• 如果需要在迭代過(guò)程中對(duì)容器進(jìn)行刪除操作必須使用迭代器, 否則會(huì)拋出ConcurrentModificationException.
• 使用迭代器刪除列表元素:

ArrayList<String> list = new ArrayList<>(Arrays.asList("I", "love", "you", "too"));
Iterator<String> it = list.iterator();
while (it.hasNext()) {
    if (it.next().length > 3) {
        it.remove();
    }
}

• 使用removeIf() 方法結(jié)合匿名內(nèi)部類(lèi)實(shí)現(xiàn):

ArrayList<String> list = new ArrayList<>(Arrays.asList("I", "love", "you", "too"));
list.removeIf(new Predicate<String>(){
    @Override
    public boolean test(String str) {
        return str.length() > 3;
    }
});

• 使用removeIf結(jié)合Lambda表達(dá)式實(shí)現(xiàn):

Array<String> list = new ArrayList<>(Arrays.asList("I", "love", "you", "too"));
list.removeIf(str -> str.length() > 3);

使用Lambda表達(dá)式不需要記憶Predicate接口名,也不需要記憶test() 方法名,只需要此處需要一個(gè)返回布爾類(lèi)型的Lambda表達(dá)式

replaceAll

• 該方法簽名: void replaceAll(UnaryOperator<E> operator);
  • 對(duì)每個(gè)元素執(zhí)行operator指定的操作,并用操作結(jié)果來(lái)替換原來(lái)的元素
    • UnaryOperator是一個(gè)函數(shù)接口,里面有待實(shí)現(xiàn)的方法T apply(T t)
• 使用下標(biāo)實(shí)現(xiàn)元素替換:

ArrayList<String> list = new ArrayList<>(Arrays.asList("I", "love", "you", "too"));
for (int i = 0; i < list.size(); i ++) {
    String str = list.get(i)
    if (str.length() > 3) {
        list.set(i, str.toUpperCase());
    }
}

• 使用replaceAll結(jié)合匿名內(nèi)部類(lèi)實(shí)現(xiàn):

ArrayList<String> list =new ArrayList<>(Arrays.asList("I", "love", "you", "too"));
list.replaceAll(new UnaryOperator<>(String){
    @Override
    public String apply(String str) {
        if (str.length() > 3) {
            return str.toUpperCase();
        }
        return str;
    }
});

代碼調(diào)用replaceAll() 方法,并使用匿名內(nèi)部類(lèi)實(shí)現(xiàn)UnaryOperator接口

• 使用Lambda表達(dá)式實(shí)現(xiàn):

ArrayList<String> list = new ArrayList<>(Arrays.asList("I", "love", "you", "too"));
list.replaceAll(str -> {
    if (str.length > 3) {
        return str.toUpperCase();
    }
    return str;
});

sort

• 該方法定義在List接口中,方法簽名: void sort(Comparator<? super E> c);
  • 根據(jù)c指定的比較規(guī)則對(duì)容器進(jìn)行排序
    • Comparator接口中需要實(shí)現(xiàn)接口int compare(T o1, T o2)
• 使用Collections的sort() 方法:

ArrayList<String> list = new ArrayList<>(Arrays.asList("I", "love", "you", "too"));
Collections.sort(list, new Comparator<String>() {
    @Override
    public int compare(String str1, String str2) {
        return str1.length() - str2.length();
    }
});

• 直接使用List.sort() 方法,結(jié)合Lambda表達(dá)式:

ArrayList<String> list = new ArrayList<>(Arrays.asList("I", "love", "you", "too"));
list.sort((str1, str2) -> str1.length() - str2.length());

spliterator

• 該方法簽名: Spliterator<E> spliterator();
  • Spliterator既可以像Iterator那樣逐個(gè)迭代,也可以批量迭代,批量迭代可以降低迭代的開(kāi)銷(xiāo)
  • Spliterator是可拆分的,一個(gè)Spliterator可以通過(guò)調(diào)用Spliterator<T> trySplit() 方法來(lái)嘗試分成兩個(gè).一個(gè)是this, 一個(gè)是新返回的元素.這兩個(gè)迭代器代表的元素沒(méi)有重疊
  • 可通過(guò)多次調(diào)用Spliterator.trySplit() 方法來(lái)分解負(fù)載,以便于多線程處理

stream和parallStream

• Stream() 和parallStream() 分別返回該容器的Stream視圖表示
• parallStream() 返回并行的Stream
• Stream是Java函數(shù)式編程的核心類(lèi)

Map

forEach

• 該方法簽名: void forEach(BiConsumer<? super K,? super V> action);
  • 對(duì)Map中的每個(gè)映射執(zhí)行action操作
    • BiConsumer是一個(gè)函數(shù)接口,里面有一個(gè)待實(shí)現(xiàn)方法 void accept(T t, U u);
• 使用Java 7之前的方式輸出Map中所有的對(duì)應(yīng)關(guān)系:

HashMap<Integer, String> map = new HashMap<>();
map.put(1, "one");
map.put(2, "two");
map.put(3, "three");
for (Map.Entry<Integer, String> entry : map.entrySet()) {
    system.out.println(entry.getKey() + "=" + entry.getValue());
}

• 使用Map的forEach() 方法,結(jié)合匿名內(nèi)部類(lèi):

HashMap<Integer, String> map = new HashMap<>();
map.put(1, "one");
map.put(2, "two");
map.put(3, "three");
map.forEach(new BiConsumer<Integer, String>() {
    @Override
    public void accept(Integer k, String v) {
        System.out.println(k + "=" + v);
    }
});

• 使用Lambda表達(dá)式:

HashMap<Integer, String> map = new HashMap<>();
map.put(1, "one");
map.put(2, "two");
map.put(3, "three");
map.forEach((k, v) -> System.out.println(k + "=" + v));

getOrDefault

• 該方法簽名: V getOrDefault(Object key, V defaultValue);
  • 按照給定的key查詢Map中對(duì)應(yīng)的value, 如果沒(méi)有找到則返回defaultValue
• 查詢Map中指定鍵所對(duì)應(yīng)的值,如果不存在則返回NoValue:

HashMap<Integer, String> map = new HashMap<>();
map.put(1, "one");
map.put(2, "two");
map.put(3, "three");
System.out.println(map.getOrDefault(4,"NoValue"));

putIfAbsent

• 該方法簽名: V putIfAbsent(K key, V value);
  • 只有在不存在key值的映射或映射值為null時(shí),才將value指定的值放入到Map中,否則不對(duì)Map做修改
  • 該方法將判斷和賦值合二為一,使用起來(lái)更加方便

remove

• 該方法簽名: remove(Object key);
  • 根據(jù)指定的key值刪除Map中映射關(guān)系
• 該方法簽名: remove(Object key, Object value);
  • 只有在當(dāng)前Map中key正好映射到value時(shí)才刪除該映射

replace

• 該方法簽名: replace(K key, V value);
  • 只有在當(dāng)前Map中key的映射存在時(shí)才用value去替換原來(lái)的值
• 該方法簽名: replace(K key, V oldValue, V newValue);
  • 只有在當(dāng)前Map中key的映射存在且等于oldValue時(shí),才用newValue去替換原來(lái)的值,否則不做任何操作

replaceAll

• 該方法簽名: replaceAll(BiFunction<? super K, ? super V, ? extends V> function);
  • 對(duì)Map中的每個(gè)映射執(zhí)行function操作,并用function的執(zhí)行結(jié)果替換原來(lái)的value
  • 其中BiFunction是一個(gè)函數(shù)接口,里面有一個(gè)待實(shí)現(xiàn)的方法R apply(T t, U u)
• 使用Java 7以前的方式將Map中的映射關(guān)系的單詞都轉(zhuǎn)換成大寫(xiě):

HashMap<Integer, String> map = new HashMap<>();
map.put(1, "one");
map.put(2, "two");
map.put(3, "three");
for (Map.Entry<Integer, String> entry : map.entrySet()) {
    entry.setValue(entry.getValue().toUpperCase());
}

• 使用replaceAll方法結(jié)合匿名內(nèi)部類(lèi):

HashMap<Integer, String> map = new HashMap<>();
map.put(1, "one");
map.put(2, "two");
map.put(3, "three");
map.replaceAll(new BiFunction<Integer, String, String>(){
    @Override
    public String apply(Integer k, String v) {
        return v.toUpperCase();
    }
});

• 使用Lambda表達(dá)式實(shí)現(xiàn):

HashMap<Integer, String> map = new HashMap<>();
map.put(1, "one");
map.put(2, "two");
map.put(3, "three");
map.replaceAll(<k, v> -> v.toUpperCase());

merge

• 該方法簽名: merge(K key, V value, BiFunction<? super V, ? super V, ? extends V> remappingFunction);
  • 如果Map中的key對(duì)應(yīng)的映射不存在或者為null, 則將value, value不可能為null關(guān)聯(lián)到key上
  • 否則執(zhí)行remappingFunction, 如果執(zhí)行結(jié)果非null, 則用該結(jié)果與key關(guān)聯(lián),否則在Map中刪除key的映射
  • 其中BiFunction是一個(gè)函數(shù)接口,里面有一個(gè)待實(shí)現(xiàn)方法R apply(T t, U u)
• merge()方法語(yǔ)義復(fù)雜,但使用的方式明確,經(jīng)典的使用場(chǎng)景: 將新的錯(cuò)誤信息拼接到原來(lái)的信息上:

map.merge(key, newMsg, (v1, v2) -> v1 + v2);

compute

• 該方法簽名: compute(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction);
  • 將remappingFunction計(jì)算的結(jié)果關(guān)聯(lián)到key上,如果計(jì)算結(jié)果為null, 則在Map中刪除key的映射
• 使用compute實(shí)現(xiàn)將新的錯(cuò)誤信息拼接到原來(lái)的信息上:

map.compute(key, (k, v) -> v == null ? newMsg : v.concat(newMsg));

computeIfAbsent

• 該方法簽名: V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction);
  • 只有在當(dāng)前Map中不存在key值的映射或映射值為null時(shí),才調(diào)用mappingFunction, 并在mappingFunction執(zhí)行結(jié)果非null時(shí),將結(jié)果跟key關(guān)聯(lián)
  • Function是一個(gè)函數(shù)接口,里面有待實(shí)現(xiàn)方法R apply(T t)
• computeIfAbsent() 常用來(lái)對(duì)Map的某個(gè)key值建立初始化映射.比如在實(shí)現(xiàn)一個(gè)多值映射時(shí) ,Map的定義可能是Map< K, Set< V > >, 要向Map中插入新值:

Map<Integer, Set<String>> map = new HashMap<>();
if (map.containsKey(1)) {
    map.get(1).add("one");
} else {
    Set<String> valueSet = new HashSet<String>();
    valueSet.add("one");
    map.put(1, valueSet);
}

• 使用Lambda表達(dá)式實(shí)現(xiàn):

Map<Integer, Set<String>> map = new HashMap<>();
map.computeIfAbsent(1, v -> new HashSet<String>()).add("one");

computeIfPresent

• 該方法簽名: V computeIfPresent(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction);
  • 作用與computeIfAbsent() 相反
  • 只有當(dāng)前Map中存在key值的映射且非null時(shí),才調(diào)用remappingFunction, 如果remappingFunction執(zhí)行結(jié)果為null, 則刪除key的映射,否則使用該結(jié)果替換key原來(lái)的映射
• Java7之前的等效代碼:

if (map.get(key) != null) {
    V oldValue = map.get(key);
    V newValue = remappingFunction.apply(key, oldValue);
    if (newValue !=null) {
        map.put(key, newValue);
    } else {
        map.remove(key);
    }
    return newValue;
}
return null;

Streams API

• stream:
  • Java函數(shù)式編程主角
  • stream不是某種數(shù)據(jù)結(jié)構(gòu),只是一種數(shù)據(jù)源視圖
  • 這里的數(shù)據(jù)源可以是:
    • 數(shù)組
    • Java容器
    • I/O channel
• stream是一個(gè)數(shù)據(jù)源視圖,需要調(diào)用對(duì)應(yīng)的工具方法創(chuàng)建一個(gè)stream:
  • 調(diào)用Collection.stream() 方法
  • 調(diào)用Collection.parallelStream() 方法
  • 調(diào)用Arrays.stream(T[] array) 方法
• stream接口繼承關(guān)系:
  
  在這里插入圖片描述
• 圖中4種stream接口繼承自BaseStream:
  • IntStream,LongStream,DoubleStream對(duì)應(yīng)三種基本類(lèi)型int, long, double. 不是對(duì)應(yīng)相應(yīng)的包裝類(lèi)型
  • Stream對(duì)應(yīng)所有剩余類(lèi)型的stream視圖
• 為不同的數(shù)據(jù)類(lèi)型設(shè)置不同的stream接口:
  • 提高性能
  • 增加特定接口函數(shù)
• 盡管stream是容器調(diào)用Collection.stream()方法得到的. stream和collections有以下不同點(diǎn):
  • 無(wú)存儲(chǔ): stream不是一種數(shù)據(jù)結(jié)構(gòu),只是數(shù)據(jù)源的一個(gè)視圖.數(shù)據(jù)源可以是數(shù)組,Java容器或I/O channel等
  • 函數(shù)式編程: 對(duì)stream的修改都不會(huì)修改背后的數(shù)據(jù)源:比如對(duì)stream執(zhí)行過(guò)濾操作并不會(huì)刪除被過(guò)濾的元素,而是會(huì)產(chǎn)生一個(gè)不包含過(guò)濾元素的新stream
  • 惰式執(zhí)行: stream上的操作不會(huì)立即執(zhí)行,只有等到真正需要stream執(zhí)行的結(jié)果時(shí)才會(huì)執(zhí)行
  • 可消費(fèi)性: stream只能被消費(fèi)一次,一旦遍歷過(guò)就會(huì)失效,就像容器的迭代器那樣,想要再次遍歷必須重新生成
• 對(duì)stream的操作分為兩類(lèi):
  • 中間操作: intermediate operations. 中間操作總會(huì)惰式執(zhí)行. 調(diào)用中間操作只會(huì)生成一個(gè)標(biāo)記了該操作的新stream
  • 結(jié)束操作: terminal operations. 結(jié)束操作會(huì)觸發(fā)實(shí)際運(yùn)算. 計(jì)算發(fā)生時(shí),會(huì)把中間積攢的操作以pipeline的方式執(zhí)行,這樣可以減少迭代操作.計(jì)算完成之后stream就會(huì)失效
• Stream接口常用方法:
  • 中間操作:
    • concat()
    • distinct()
    • filter()
    • flatMap()
    • limit()
    • map()
    • peek()
    • skip()
    • sorted()
    • parallel()
    • sequential()
    • unordered()
  • 結(jié)束操作:
    • allMatch()
    • anyMatch()
    • collect()
    • count()
    • findAny()
    • findFirst()
    • forEach()
    • forEachOrdered()
    • max()
    • min()
    • noneMatch()
    • reduce()
    • toArray()
• 區(qū)分中間操作和結(jié)束操作就是看方法的返回值:
  • 返回值為stream的大都是中間操作
  • 否則是結(jié)束操作
• Stream方法使用:
  • stream與函數(shù)接口關(guān)系非常緊密,沒(méi)有函數(shù)接口stream就無(wú)法操作
    • 函數(shù)接口是指內(nèi)部只有一個(gè)抽象方法的接口
    • 函數(shù)接口出現(xiàn)的地方都可以使用Lambda表達(dá)式

forEach

• 該方法簽名: void forEach(Consumer<? super E> action);
  • 對(duì)容器中的每個(gè)元素執(zhí)行action指定的操作,即對(duì)元素進(jìn)行遍歷

/* 
 * 使用Stream.forEach()進(jìn)行迭代
 */
 Stream<String> stream = Stream.of("I", "love", "you", "too");
 stream.forEach(str -> System.out.println(str)); 

• 由于forEach() 是結(jié)束方法,所以上述方法會(huì)立即執(zhí)行,輸出所有字符串

filter

在這里插入圖片描述

• 該函數(shù)原型: Stream< T > filter(Predicate<? super T> predicate);
  • 返回一個(gè)只包含滿足predicate條件元素的stream

/*
 * 保留長(zhǎng)度等于3的字符串
 */
 Stream<String> stream = Stream.of("I", "love", "you", "too");
 stream.filter(str -> str.length() == 3)
       .forEach(str -> System.out.println(str));

• 輸出長(zhǎng)度等于3的字符串you和too
• 由于filter() 是個(gè)中間操作, 如果只調(diào)用filter() 不會(huì)有實(shí)際計(jì)算,因此不會(huì)輸出任何信息

distinct

在這里插入圖片描述

• 該函數(shù)原型: Stream< T > distinct();
  • 返回一個(gè)去除重復(fù)元素之后的Stream

Stream<String> stream = Stream.of("I", "love", "you", "too", "too");
stream.distinct()
      .forEach(str -> System.out.println(str));

• 輸出去掉一個(gè)too之后的其余字符串

sorted

• 排序函數(shù)有兩個(gè):
  • 自然順序排序: Stream< T > sorted();
  • 使用自定義比較器排序: Stream < T > sorted(Comparator<? super T> comparator);

Stream<String> stream = Stream.of("I", "love", "you", "too");
stream.sorted((str1, str2) -> str1.length() - str2.length())
      .forEach(str -> System.out.println(str));

• 輸出按照長(zhǎng)度升序排序后的字符串

map

在這里插入圖片描述

• 該函數(shù)原型: < R > Stream< R > map(Function<? super T, ? extends R> mapper);
  • 返回一個(gè)對(duì)當(dāng)前所有元素執(zhí)行mapper之后的結(jié)果組成的Stream
  • 就是對(duì)每個(gè)元素按照某種操作進(jìn)行轉(zhuǎn)換 ,轉(zhuǎn)換前后Stream中元素的個(gè)數(shù)不會(huì)改變,但是元素的類(lèi)型取決于轉(zhuǎn)換之后的類(lèi)型

Stream<String> stream = Stream.of("I", "love", "you", "too");
stream.map(str -> str.toUpperCase())
      .forEach(str -> System.out.println(str));

• 輸出原字符串的大寫(xiě)形式

flatMap

在這里插入圖片描述

• 該函數(shù)原型: < R > Stream< R > flatMap(Function<? super T, ? extends Stream<? extends R>> mapper);
  • 對(duì)每一個(gè)元素執(zhí)行mapper的指定操作,并用所有mapper返回的Stream中的元素作為一個(gè)新的Stream作為最終返回結(jié)果
  • 就相當(dāng)于將原Stream中的所有元素都"攤平"之后組成的新Stream. 轉(zhuǎn)換前后元素的個(gè)數(shù)和類(lèi)型都可能會(huì)改變

Stream<List<Integer>> stream = Stream.of(Arrays.asList(1,2), Arrays.asList(3,4,5));
stream.flatMap(list -> list.stream())
      .forEach(i -> System.out.println(i)); 

• 原來(lái)的stream中有兩個(gè)元素,分別是兩個(gè)List< Integer >, 執(zhí)行flatMap() 之后,將每一個(gè)List都"攤平"成一個(gè)個(gè)數(shù)字,所以會(huì)產(chǎn)生一個(gè)由5個(gè)數(shù)字組成的Stream. 所以最終將輸出1~5這5個(gè)數(shù)字

Stream API 高級(jí)

• 歸約操作: reduction operation
  • 又稱作折疊操作fold
  • 通過(guò)某個(gè)連接動(dòng)作將所有元素匯總成一個(gè)匯總結(jié)果的過(guò)程
  • 元素求和,求最大值最小值,求元素總個(gè)數(shù),將所有元素轉(zhuǎn)換成一個(gè)列表或集合,都屬于歸約操作
• Stream類(lèi)庫(kù)中兩個(gè)通用的規(guī)約操作:
  • reduce()
  • collect()
• 也有一些為了簡(jiǎn)化書(shū)寫(xiě)而設(shè)計(jì)的專(zhuān)用歸約操作 : sum(), max(), min(), count() 等

reduce

• 實(shí)現(xiàn)從一組元素中生成一個(gè)值
• sum(),max(),min(),count()等都是reduce操作,將這些單獨(dú)設(shè)為函數(shù)是因?yàn)榻?jīng)常使用
• reduce()的方法定義有三種重寫(xiě)形式:
  • Optional< T > reduce(BinaryOperator< T > accumulator);
  • T reduce(T identity, BinaryOperator< T > accumulator);
  • < U > U reduce(U identity, BiFunction< U, ? super T,U> accumulator, BinaryOperator< U > combiner);
  • 雖然函數(shù)的定義越來(lái)越長(zhǎng),但是語(yǔ)義不變.多的參數(shù)是為了指明初始值(identity), 或者是指定并行執(zhí)行時(shí)多個(gè)部分結(jié)果的合并方式(combiner)
• 從一組單詞中找出最長(zhǎng)的單詞.這里"大"的含義就是"長(zhǎng)":

  /*
   * 找出最長(zhǎng)的單詞
   */
  Stream<String> stream = Stream.of("I", "love", "you", "too");
  Optional<String> longets = stream.reduce((s1, s2) -> s1.length() >= s2.length() ? s1 : s2);
  // Optional<String> longest = stream.max((s1, s2) -> s1.length() - s2.length());
  System.out.println(longest.get());
  

  • 選出最長(zhǎng)的單詞love
  • 其中Optional是只有一個(gè)值的容器,使用Optional可以避免null值

• 求出一組單詞長(zhǎng)度之和. 這是個(gè)求和操作,操作對(duì)象輸入類(lèi)型是String,結(jié)果類(lèi)型是Integer:

  
  
  在這里插入圖片描述

/*
 * 求單詞長(zhǎng)度之和
 */
 Stream<String> strean = Stream.of("I", "love", "you", "too");
 Integer lengthSum = stream.reduce(0,       // 初始值 (1)
                                  (sum, str) -> sum +str.length(),      // 累加器 (2)
                                  (a, b) -> a + b);     // 部分和拼接,并行執(zhí)行時(shí)會(huì)用到 (3)
// int lengthSun = stream.mapToInt(str -> str.length()).sum();
System.out.println(lengthSum);

• 上述代碼 (2) 處的累加器:
  • 字符串映射成長(zhǎng)度
  • 并和當(dāng)前累加和相加
  • 使用reduce() 函數(shù)將這兩步合二為一,更有助于提升性能
• 同樣也可以使用map() 和sum() 組合也可以達(dá)到目的

collect

• reduce() 的優(yōu)點(diǎn)的是生成一個(gè)值,但是如果想要從Stream中生成一個(gè)集合或者Map等復(fù)雜對(duì)象時(shí),就要用到collect()
• 示例:

/* 
 * 將Stream轉(zhuǎn)換成容器或者M(jìn)ap
 */
Stream<String> stream = Stream.of("I", "love", "you", "too");
List<String> list = stream.collect(Collectors.toList());
Set<String> set = stream.collect(Collectors.toSet());
// 將Stream轉(zhuǎn)換成Map
Map<String, Integer> map = stream.collect(Collectors.toMap(Function.identity(), String::length));

• 上述分別將Stream轉(zhuǎn)換成List,Set,Map
• 需要注意的有:
  • Function.identity()
  • String::length
  • Collectors

接口的靜態(tài)方法和默認(rèn)方法

• Function是一個(gè)接口 ,Function.identity() 含義有兩個(gè)方面:
  • Java 8允許在接口中加入具體方法. 接口中的具體方法有兩種:
    • static: 靜態(tài)方法,identity()就是Function接口的一個(gè)靜態(tài)方法
    • default: 默認(rèn)方法
  • Function.identity(): 返回一個(gè)輸出和輸入一樣的Lambda表達(dá)式對(duì)象,等價(jià)于 t -> t 形式的Lambda表達(dá)式

在Java 7之前要想在定義好的接口中加入新的抽象方法是很困難甚至不可能的,因?yàn)闀?huì)所有實(shí)現(xiàn)了該接口的類(lèi)都要重新實(shí)現(xiàn).Java 8中的default方法就是用來(lái)解決這個(gè)問(wèn)題,直接在接口中實(shí)現(xiàn)新加入的方法,引進(jìn)了default方法之后,可以繼續(xù)加入static方法來(lái)避免專(zhuān)門(mén)的工具類(lèi)

方法引用

• 形如String::length的語(yǔ)法格式叫作方法引用(method reference),這種語(yǔ)法用來(lái)替代某些特定形式的Lambda表達(dá)式
• 如果Lambda表達(dá)式的全部?jī)?nèi)容就是調(diào)用一個(gè)已有的方法,就可以用方法引用來(lái)代替Lambda表達(dá)式
• 方法引用可以分為四類(lèi):
  • 引用靜態(tài)方法: Integer :: sum
  • 引用某個(gè)對(duì)象的方法: list :: add
  • 引用某個(gè)類(lèi)的方法: String :: length
  • 引用構(gòu)造方法: HashMap :: new

Collector

在這里插入圖片描述

• 收集器Collector是為Stream.collect方法打造的工具類(lèi)
• 將一個(gè)Stream轉(zhuǎn)換成一個(gè)容器或者M(jìn)ap至少需要考慮兩個(gè)方面:
  • 目標(biāo)容器是什么: ArrayList,HashSet還是TreeMap
  • 新元素如何添加到目標(biāo)容器中: List.add()還是Map.put()
  • 如果是并行進(jìn)行規(guī)約,還要使得collect()如何做到將多個(gè)部分結(jié)果合并成一個(gè)
• collect() 方法定義: < R > R collect(Supplier< R > supplier, BiConsumer<R, ? super T> accumulator, BiConsumer<R, R> combiner);
  • 三個(gè)參數(shù)一次對(duì)應(yīng)著上面的三條分析
  • 每次調(diào)用collect() 都要傳入這三個(gè)參數(shù)很麻煩,所以使用收集器Collector對(duì)三個(gè)參數(shù)進(jìn)行簡(jiǎn)單的封裝
  • 所以collect() 另一個(gè)方法定義: < R, A> R collect(Collecor<? super T, A, R> collector);
• Collectors工具類(lèi)可以通過(guò)靜態(tài)方法生成各種常用的Collector
• 這樣,將Stream規(guī)約成List可以通過(guò)如下兩種方式:

/*
 * 將Stream規(guī)約成List
 */
 Stream<String> stream = Stream.of("I", "love", "you", "too");
 
 List<String> list1 = stream.collect(ArrayList :: new, ArrayList :: add, ArrayList :: addAll);
 System.out.println(list1);

 List<String> list2 = stream.collect(Collectors.toList());
 System.out.println(list2);

• 通常情況下不需要手動(dòng)指定collect() 的三個(gè)參數(shù),而是調(diào)用collect(Collector<? super T, A, R> collector) 方法,并且參數(shù)中的Collector對(duì)象大都是直接通過(guò)Collectors工具類(lèi)獲得
• 實(shí)際傳入的收集器的行為決定collect()的行為

使用collect()生成Collection

• 通過(guò)collect() 方法將Stream轉(zhuǎn)換成容器的方法中將Stream轉(zhuǎn)換成List和Set是最常見(jiàn)的操作
• 在Collectors工具類(lèi)中已經(jīng)提供了對(duì)應(yīng)的收集器:

/*
 * 將Stream轉(zhuǎn)換成List或者Set
 */
 Stream<String> stream = Stream.of("I", "love", "you", "too");
 
 List<String> list = stream.collect(Collectors.toString());
 Set<String> set = stream.collect(Collectors.toSet());

• 由于返回結(jié)果是接口類(lèi)型,所以并不清楚類(lèi)庫(kù)實(shí)際選擇的容器類(lèi)型什么
• 有時(shí)需要人為指定容器的實(shí)際類(lèi)型,這個(gè)需求可以通過(guò)Collectors.toCollection(Supplier< C > collectionFactory) 完成:

/*
 * 使用toCollection指定規(guī)約容器的類(lèi)型
 */
 ArrayList<String> arrayList = stream.collect(Collectors.toCollection(ArrayList :: new));
 HashSet<String> hashSet = stream.collect(Collectors.toCollection(HashSet :: new));

• 分別指定規(guī)約結(jié)果為ArrayList和HashSet

使用collect()生成Map

• Stream依賴某種數(shù)據(jù)源,數(shù)據(jù)源可以是數(shù)組,容器等,但不能是Map
• 但是可以從Stream生成Map,要做的是確定好Map的key和value分別代表什么,這個(gè)在于要想清楚到底要干什么
• 通常在三種情況下collect()的結(jié)果會(huì)是Map:
  • 使用Collectors.toMap() 生成的收集器: 用戶需要指定如何生成Map的key和value
  • 使用Collectors.partitioningBy() 生成的收集器: 對(duì)元素進(jìn)行二分區(qū)操作時(shí)用到
  • 使用Collectors.groupingBy() 生成的收集器: 對(duì)元素做group操作時(shí)用到
• 使用toMap()生成的收集器:
  • 這個(gè)是和Collectors.toCollection() 并列的方法
  • 示例: 將學(xué)生列表轉(zhuǎn)換成由<學(xué)生, GPA>組成的Map

/*
 * 使用toMap()統(tǒng)計(jì)學(xué)生的GPA
 */
 Map<Student, Double> studentToGPA = student.stream().collect(Collectors.toMap(Function.identity(), // 如何生成key
                                                                                student -> computeGPA(student)));   // 如何生成value                                                                

• 使用partitioningBy()生成的收集器:
  • 適用于將Stream中的元素依據(jù)某個(gè)二值邏輯(Boolean 滿足,不滿足)分成互補(bǔ)相交的兩部分
  • 示例: 將學(xué)生分成成績(jī)及格和不及格的兩部分

/*
 * 將學(xué)生成績(jī)分為及格不及格兩部分
 */
 Map<Boolean, List<Student>> passingFailing = students.stream().collect(Collectors.partitioningBy(s -> s.getGrade() >= PASS_THRESHOLD));

• 使用groupingBy()生成的收集器:
  • 這是比較靈活的一種,與SQL中的group by語(yǔ)句類(lèi)似
  • 這里的groupingBy也是按照某個(gè)屬性對(duì)數(shù)據(jù)進(jìn)行分組,屬性相同的元素會(huì)被對(duì)應(yīng)到Map的同一個(gè)key上
  • 示例: 將員工按照部門(mén)進(jìn)行分組

/*
 * 將員工按照部門(mén)進(jìn)行分組
 */
 Map<Department, List<Employee>> byDept = employees.stream().collect(Collectors.groupingBy(Employee :: getDepartment));

• 有時(shí)候僅僅分組是無(wú)法滿足要求的.在SQL中使用group by是為了方便更高級(jí)的查詢:
  • 先將員工按照部門(mén)分組
  • 然后統(tǒng)計(jì)每個(gè)部門(mén)員工的人數(shù)
  • 增強(qiáng)版的groupingBy()能夠滿足這種需求:
    • 增強(qiáng)版的groupingBy()允許先對(duì)元素分組之后再執(zhí)行某種運(yùn)算,比如求和,計(jì)數(shù),平均值,類(lèi)型轉(zhuǎn)換等
    • 這種先將元素分組的收集器叫作上游收集器
    • 然后執(zhí)行分組后的運(yùn)算的收集器叫作下游收集器

/*
 * 使用下游收集器統(tǒng)計(jì)每個(gè)部門(mén)的人數(shù)
 */
 Map<Department, Integer> totalByDept = employees.stream()
                                                 .collect(Collectors.groupingBy(Employee :: getDepartment,
                                                                                Collectors.counting()));

這個(gè)groupingBy和SQL相似,也是高度非結(jié)構(gòu)化

• 下游收集器還可以包含更下游的收集器:
  • 將員工按照部門(mén)分組
  • 得到每個(gè)員工的名字字符串,而不是一個(gè)個(gè)Employee對(duì)象

/*
 * 按照部門(mén)對(duì)員工進(jìn)行分組,并且只保留員工的名字
 */
 Map<Department, List<String>> byDept = employees.stream()
                                                 .collect(Collectors.groupingBy(Employee :: getDepartment,
                                                  Collectors.mapping(Employee :: getName,
                                                                    Collectors.toList())));

使用collect()做字符串join

• 字符串拼接時(shí)使用Collectors.joining() 生成的收集器,代替for循環(huán)拼接
• Collectors.joining() 方法有三種重寫(xiě)形式,分別對(duì)應(yīng)三種不同的拼接方式:

/*
 * 使用Collectors.joining()拼接字符串
 */
 Stream<String> stream = Stream.of("I", "love", "you");

String joined = stream.collect(Collectors.joining());   // Iloveyou
String joined = stream.collect(Collectors.joining(","));    // I,love,you
String joined = stream.collect(Collectors.joining(",", "{", "}"));  // {I,love,you}

• 除了可以使用Collectors工具類(lèi)已經(jīng)封裝好的收集器,還可以自定義收集器.或者直接調(diào)用collect(Supplier< R > supplier, BiConsumer<R, ? super T> accumulator, BiConsumer<R, R> combiner) 方法,收集需要的任何形式的信息

Stream Pipelines

• 通過(guò)使用Stream API中引起的疑問(wèn):
  • 如此強(qiáng)大的Stream API是如何實(shí)現(xiàn)的?
  • Pipeline是怎么執(zhí)行的,每次調(diào)用都會(huì)迭代一次嗎?
  • 自動(dòng)并行又是怎么做到的,線程個(gè)數(shù)是多少?
• 容器執(zhí)行Lambda表達(dá)式的方式 - 以ArrayList.forEach()方法為例:

/*
 * ArrayList.forEach()
 */
 public void forEach(Consumer<? super E> action) {
    ...
    for (int i = 0; modCount == expectedModCount && i < size; i ++) {
        // 回調(diào)方法
        action.accept(elementData[i]);
    }
    ...
 }

• ArrayList.forEach() 方法的主要邏輯就是一個(gè)for循環(huán),在該for循環(huán)里不斷調(diào)用action.accept() 回調(diào)方法完成對(duì)元素的遍歷
• 回調(diào)方法在Java GUI的監(jiān)聽(tīng)器中廣泛使用,Lambda表達(dá)式的作用就是相當(dāng)于一個(gè)回調(diào)方法
• Stream API中大量使用Lambda表達(dá)式作為回調(diào)方法. 但想要理解Stream, 關(guān)鍵的是:
  • 流水線
  • 自動(dòng)并行

int longestStringLengthStaringWithA = strings.stream().filter(s -> s.startsWith("A"))
                                                      .mapToInt(String :: length)
                                                      .max();

• 上述代碼用來(lái)求出以字母 "A" 開(kāi)頭的字符串的最大長(zhǎng)度:
  • 一種直白的方式就是為每一次函數(shù)調(diào)用都執(zhí)行一次迭代. 盡管這樣做能夠?qū)崿F(xiàn)功能,但效率上是無(wú)法接受的
  • 類(lèi)庫(kù)的實(shí)現(xiàn)是使用Stream Pipeline的方式巧妙地避免了多次迭代.基本思想就是在一次迭代中盡可能多的執(zhí)行用戶指定的操作
• Stream中的相關(guān)操作:
  • 中間操作: Intermediate operations
    • 無(wú)狀態(tài): Stateless
      • unordered()
      • filter()
      • map()
      • mapToInt()
      • mapToLong()
      • mapToDouble()
      • flatMap()
      • flatMapToInt()
      • flatMapToLong()
      • flatMapToDouble()
      • peek()
    • 有狀態(tài): Stateful
      • distinct()
      • sorted()
      • limit()
      • skip()
  • 結(jié)束操作: Terminal operations
    • 短路操作: short-circuiting
      • anyMatch()
      • allMatch()
      • noneMatch()
      • findFirst()
      • findAny()
    • 非短路操作:
      • forEach()
      • forEachOrdered()
      • toArray()
      • reduce()
      • collect()
      • max()
      • min()
      • count()

• Stream上的所有操作分為兩類(lèi): 因?yàn)镾tream底層對(duì)每一種情況的處理方式不同,所以要進(jìn)行精細(xì)的劃分
  • 中間操作: 中間操作只是一種標(biāo)記
    • 無(wú)狀態(tài): 指元素的處理不受前面元素的影響,處理完一個(gè)元素就能立即知道結(jié)果
    • 有狀態(tài): 指元素的處理受到別的元素的影響,必須等到所有元素處理之后才能知道結(jié)果
  • 結(jié)束操作: 只有結(jié)束操作才會(huì)觸發(fā)實(shí)際的計(jì)算
    • 短路操作: 指不用處理全部元素就可以返回結(jié)果
    • 非短路操作: 指對(duì)所有的元素處理后才可以返回結(jié)果

Stream Pipeline實(shí)現(xiàn)方案

• 一種直白的Stream Pipeline實(shí)現(xiàn)方案:

  
  
  在這里插入圖片描述
• 求最長(zhǎng)字符串的長(zhǎng)度:
  • 一種直白的實(shí)現(xiàn)方式是為每一次函數(shù)調(diào)用都執(zhí)行一次迭代,并將處理中間結(jié)果發(fā)明放到某種數(shù)據(jù)結(jié)構(gòu)中,比如數(shù)組,容器等
    • 就是調(diào)用filter() 方法后立即執(zhí)行
    • 選出所有以A開(kāi)頭的字符串并放到一個(gè)列表list1中
    • 然后讓list1傳遞給mapToInt() 方法并立即執(zhí)行
    • 生成的結(jié)果放到list2中
    • 最后遍歷list2, 找出最大的數(shù)字作為最終的結(jié)果
  • 這種實(shí)現(xiàn)方法實(shí)現(xiàn)簡(jiǎn)單直觀,但存在兩個(gè)明顯的缺陷:
    • 迭代次數(shù)多: 迭代次數(shù)和函數(shù)的調(diào)用次數(shù)相等
    • 頻繁產(chǎn)生中間結(jié)果: 每次函數(shù)調(diào)用都產(chǎn)生一次中間結(jié)果,存儲(chǔ)開(kāi)銷(xiāo)大
• 不使用Stream API在一次迭代中實(shí)現(xiàn)求最長(zhǎng)字符串長(zhǎng)度的方式:

int longest = 0;
for (String str : strings) {
    if (str.startsWith("A")) {  // 類(lèi)似filter(),保留以A開(kāi)頭的字符串
        int len = str.length(); // 類(lèi)似mapToInt(),得到字符串的長(zhǎng)度
        longest = Math.max( );
    }
}

• 采用這種方法不但減少了迭代次數(shù),也避免了存儲(chǔ)中間結(jié)果,這就是Stream Pipeline.將三個(gè)操作放在了一次迭代中
  • 只要事先知道意圖,總是能夠采取上述方式實(shí)現(xiàn)與Stream API等價(jià)的功能

Stream Pipeline解決方法

• 由于Stream類(lèi)庫(kù)的設(shè)計(jì)者不知道用戶意圖,所以如何在無(wú)法假設(shè)用戶行為的前提下,是類(lèi)庫(kù)的設(shè)計(jì)者要考慮的問(wèn)題?
• 關(guān)于這個(gè)解決方法,可以采用某種方式記錄用戶每一步的操作,當(dāng)用戶調(diào)用結(jié)束操作時(shí)將之前記錄的操作疊加到一起在一次迭代中全部執(zhí)行完成. 關(guān)于這種解決方法,需要解決以下四個(gè)問(wèn)題:
  • 用戶的操作如何記錄?
  • 操作如何疊加?
  • 疊加之后的操作如何執(zhí)行?
  • 執(zhí)行后的結(jié)果在哪里?

操作如何記錄?

• 這里的操作指的是Stream中間操作
• 很多Stream的操作會(huì)需要一個(gè)回調(diào)函數(shù) - Lambda表達(dá)式,因此一個(gè)完整的操作應(yīng)該是一個(gè)三元數(shù)組:
  • <數(shù)據(jù)來(lái)源, 操作, 回調(diào)函數(shù)>
• Stream中使用Stage的概念來(lái)描述一個(gè)完整的操作,并用某種實(shí)例化后的PipelineHelper來(lái)代表Stage, 將具有先后順序的各個(gè)Stage連到一起,就構(gòu)成了整個(gè)Stream Pipeline
• Stream相關(guān)類(lèi)和接口的繼承關(guān)系圖:
  
  image
• IntPipeline.LongPipeline,DoublePipeline三個(gè)類(lèi)是專(zhuān)門(mén)為三種基本類(lèi)型而不是包裝類(lèi)型定制的,與ReferencePipeline是并列關(guān)系
• 圖中Head用于表示第一個(gè)Stage, 即調(diào)用諸如Collection.stream() 方法產(chǎn)生的Stage, 很顯然這個(gè)Stage中不包含任何操作
• StatelessOp和StatefulOp分別表示無(wú)狀態(tài)和有狀態(tài)的Stage, 對(duì)應(yīng)于無(wú)狀態(tài)和有狀態(tài)的操作
• Stream Pipeline組織結(jié)構(gòu)示意圖:
  
  -
• 通過(guò)Collection.stream() 方法得到Head, 即Stage0, 緊接著調(diào)用一系列中間操作,不斷產(chǎn)生新的Stream
• 這些Stream對(duì)象以雙向鏈表的形式組織在一起,構(gòu)成整個(gè)流水線,由于每個(gè)Stage都記錄了前一個(gè)Stage和本次操作的回調(diào)函數(shù),依靠這種數(shù)據(jù)結(jié)構(gòu)就能建立起對(duì)所有數(shù)據(jù)源的操作. 這就是Stream記錄操作的方式

操作如何疊加?

• 通過(guò)上面的方法解決了操作記錄問(wèn)題,要想讓Stream Pipeline起到應(yīng)有的作用需要一種將所有操作疊加到一起的方案
• 因?yàn)橹挥挟?dāng)前Stage本身才知道該如何執(zhí)行自己包含的操作.前面的Stage并不知道后面的Stage到底執(zhí)行了哪種操作,以及回調(diào)函數(shù)是哪種形式,所以不能夠從Stream Pipeline的Head開(kāi)始依次執(zhí)行每一步操作與回調(diào)函數(shù)來(lái)實(shí)現(xiàn)操作疊加
• 為了解決以上問(wèn)題,就需要某種協(xié)議來(lái)協(xié)調(diào)相鄰Stage之間的調(diào)用關(guān)系, 這種協(xié)議由Sink接口完成 ,Sink接口包含的方法如下所示:

• 通過(guò)Sink協(xié)議,可以方便地進(jìn)行相鄰的Stage調(diào)用,每個(gè)Stage都會(huì)將自己的操作封裝到一個(gè)Sink里,前一個(gè)Stage只需要調(diào)用最后一個(gè)Stage的accept() 方法即可,并不需要知道Stage的內(nèi)部是如何處理的
• 對(duì)于有狀態(tài)的操作,Sink的begin() 和end() 方法是必須實(shí)現(xiàn)的:
  • 比如Stream.sorted() 是一個(gè)有狀態(tài)的中間操作
  • 對(duì)應(yīng)的Sink.begin() 方法可能創(chuàng)建一個(gè)存放結(jié)果的容器
  • 而accept() 方法負(fù)責(zé)將元素添加到容器中
  • 最后end() 負(fù)責(zé)對(duì)容器進(jìn)行排序
• 對(duì)于短路操作,Sink.cancellationRequest() 是必須實(shí)現(xiàn)的:
  • 比如Stream.findFirst() 是一個(gè)短路操作
  • 只要找到一個(gè)元素 ,cancellationRequested() 就應(yīng)該返回true, 以便調(diào)用者盡快結(jié)束查找
• Sink的四個(gè)接口方法互相協(xié)作,共同完成計(jì)算任務(wù). 實(shí)際上Stream API內(nèi)部實(shí)現(xiàn)的本質(zhì),就是如何重載Sink的這四個(gè)接口方法
• 根據(jù)Sink對(duì)操作的包裝,就解決了Stage之間的調(diào)用問(wèn)題,執(zhí)行時(shí)只需要從流水線的head開(kāi)始對(duì)數(shù)據(jù)源依次調(diào)用每個(gè)Stage對(duì)應(yīng)的Sink.{begin(), accept(), cancellationRequested(), end()} 方法
• 示例: 一種可能的Sink.accept()方法流程

void accept(U u) {
    1. 使用當(dāng)前Sink包裝的回調(diào)函數(shù)處理u
    2. 將處理結(jié)果傳遞給Pipeline下游的Sink
}

• Sink接口的方法都是按照 [處理 -> 轉(zhuǎn)發(fā)] 的模型實(shí)現(xiàn)
• 示例: Stream的中間操作是如何將自身的操作包裝成Sink以及Sink如何將結(jié)果轉(zhuǎn)發(fā)給下一個(gè)Sink的,Stream.map()方法如下

/*
 * Stream.map(),調(diào)用該方法將產(chǎn)生一個(gè)新的Stream
 */
public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {
    ...
    return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE, StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
        @override
        /*
         * opWripSink()方法返回由回調(diào)函數(shù)包裝成的Sink
         */
         Sink<P_OUT> opWrapSink(int flags, Sink<R> downStream) {
            return new Sink.ChainedReference<P_OUT, R>(downStream) {
                @Override
                public void accept(P_OUT u) {
                    // 使用當(dāng)前Sink包裝的回調(diào)函數(shù)mapper處理u
                    R r = mapper.apply(u)
                    // 將處理結(jié)果傳遞給流水線下游的Sink
                    downstream.accept(r);
                }
            };
         }
    };
}

• 將回調(diào)函數(shù)mapper包裝到一個(gè)Sink中:
  • Stream.map() 是一個(gè)無(wú)狀態(tài)的中間操作,所以map() 方法返回了一個(gè)StatelessOp內(nèi)部類(lèi)對(duì)象,一個(gè)新的Stream
  • 調(diào)用這個(gè)新Stream的opWripSink() 方法將得到一個(gè)包裝了當(dāng)前回調(diào)函數(shù)的Sink
• 示例:
  • Stream.sorted() 方法將對(duì)Stream中的元素進(jìn)行排序
  • 這是一個(gè)有狀態(tài)的中間操作,因?yàn)樵谧x取所有元素之前是無(wú)法獲得最終順序的
  • sorted()方法封裝的Sink如下:

/*
 * Stream.sort()中的Sink實(shí)現(xiàn)
 */
 class RefSortingSink<T> extends AbstractRefSortingSink<T> {
    // 存放用于排序的元素
    private ArrayList<T> list;

    RefSortingSink(Sink<? super T> downstream, Comparator<? super T> comparator) {
        super(downstream, comparator);
    }

    @Override
    public void begin(long size) {
        ...
        // 創(chuàng)建一個(gè)存放排序元素的列表
        list = (size > 0) ? new ArrayList<T>((int)size) : new ArrayList<T>();
    }

    @Override
    public void end() {
        // 只有全部元素結(jié)束接收之后才能開(kāi)始排序
        list.sort(comparator);
        downstream.begin(list.size());
        
        if (!cacellationWasRequested()) {
            // 如果下游Sink不包含短路操作,將處理結(jié)果傳遞給流水線下游的Sink
            list.forEach(downstream :: accept);
        } else {
            /*
             * 如果下游Sink包含短路操作:
             *  每次都調(diào)用cancellationRequested()詢問(wèn)是否可以結(jié)束處理
             */
             for (T t : list) {
                if (down.cancellationWasRequested()) {
                    break;
                }
                // 將處理結(jié)果傳遞給流水線下游的Sink
                downstream.accept();
             }
        }
        downstream.end();
        list = null;
    }

    @Override
    public void accept(T t) {
        /*
         * 使用當(dāng)前Sink包裝動(dòng)作處理:
         *  將元素添加到中間列表中
         */
        list.add(t);
    }
 }

• Sink中的四個(gè)接口方法的協(xié)作方式:
  • 首先begin() 方法獲取參與排序的元素個(gè)數(shù)傳遞給Sink. 方便確定中間結(jié)果容器的大小
  • 然后通過(guò)accept() 方法將元素添加到中間結(jié)果中,最終執(zhí)行時(shí)調(diào)用者會(huì)不斷調(diào)用該方法,直到遍歷所有元素
  • 最后end() 方法返回給Sink所有元素遍歷完畢,啟動(dòng)排序步驟,排序完成后將結(jié)果傳遞給下游的Sink
  • 如果下游的Sink是短路操作,將結(jié)果傳遞給下游時(shí)不斷詢問(wèn)下游cancellationRequested() 是否可以結(jié)束處理

疊加操作如何執(zhí)行?

在這里插入圖片描述

• Sink封裝了Stream的每一步操作,并使用 [處理 -> 轉(zhuǎn)發(fā)] 的模式來(lái)疊加操作.一旦調(diào)用某個(gè)結(jié)束操作, 就會(huì)觸發(fā)整個(gè)流水線的執(zhí)行
  • 結(jié)束操作之后不會(huì)有別的操作,所以結(jié)束操作不會(huì)創(chuàng)建新的流水線階段Stage. 流水線的鏈表不會(huì)再往后延伸
  • 結(jié)束操作會(huì)創(chuàng)建一個(gè)包裝了自己操作的Sink, 這是最后一個(gè)Sink, 不會(huì)有下游的Sink. 所以這個(gè)Sink只需要處理數(shù)據(jù)而不需要將結(jié)果傳遞給下游的Sink
  • 對(duì)于Sink的 [處理 -> 轉(zhuǎn)發(fā)] 模型,結(jié)束操作的Sink就是調(diào)用鏈的出口
• 上游Sink如何找到下游Sink:
  • 一種方案是在PipelineHelper中設(shè)置一個(gè)Sink字段,在流水線中找到下游Stage并訪問(wèn)Sink字段即可
  • 在Stream中,設(shè)置了一個(gè)SinkAbstractPipeline.opWrapSink(int flag, Sink downstream) 方法來(lái)得到Sink. 該方法的作用:
    • 返回一個(gè)新的包含了當(dāng)前Stage代表的操作以及能夠?qū)⒔Y(jié)果傳遞給downstream的Sink對(duì)象
• 使用一個(gè)新的Sink對(duì)象而不是返回一個(gè)Sink字段:
  • 因?yàn)槭褂?strong>opWrapSink() 可以將當(dāng)前操作與下游Sink的downstream參數(shù)結(jié)合成新的Sink
  • 這樣只要從流水線的最后一個(gè)Stage開(kāi)始,不斷調(diào)用上一個(gè)Stage的opWrapSink() 方法直到最開(kāi)始(不包括stage0, 因?yàn)?strong>stage0代表數(shù)據(jù)源,不包含操作),就可以得到一個(gè)代表了流水線上所有操作的Sink

/**
 * AbstractPipeline.wrapStack():
 * 從下游向上游不斷包裝Sink,如果最初傳入的Sink代表結(jié)束操作,函數(shù)返回時(shí)就可以得到一個(gè)代表了流水線上所有操作的Sink
 */
final <P_IN> Sink<P_IN> wrapSink() {
    ...
    for (AbstractPipeline p = AbstractPipeline.this; p.depth > 0; p = p.previousStage) {
        sink = p.opWrapSink(p.previousStage.combinedFlags, sink);
    }
    return (Sink<P_IN>) sink;
}

• 流水線Stage上從開(kāi)始到結(jié)束的所有操作都被包裝到一個(gè)Sink里,執(zhí)行這個(gè)Sink就相當(dāng)于執(zhí)行整個(gè)流水線:

/*
 * AbstractPipeline.copyInto():
 *  對(duì)spliterator代表的數(shù)據(jù)執(zhí)行wrappedSink代表的操作
 */
 final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
    ...
    if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags)) {
        // 通知開(kāi)始遍歷履歷
        wrappedSink.begin(spliterator.getExactSizeIfKnown());
        // 迭代
        spliterator.forEachRemaining(wrappedSink);
        // 通知遍歷結(jié)束
        wrappedSink.end();
    }
    ...
 }

上述代碼首先調(diào)用wrappedSink.begin() 方法告訴Sink數(shù)據(jù)即將到來(lái),然后調(diào)用Spliterator迭代器的spliterator.forEachRemaining() 方法對(duì)數(shù)據(jù)進(jìn)行迭代,最后調(diào)用wrappedSink.end() 方法通知Sink數(shù)據(jù)處理結(jié)束

執(zhí)行后的結(jié)果位置

• 首先不是所有的Stream結(jié)束操作都需要返回結(jié)果,有些操作只是為了使用副作用Side-effects :
  • 比如Stream.forEach() 方法將結(jié)果打印出來(lái)就是常見(jiàn)的副作用場(chǎng)景
  • 事實(shí)上,除了打印之外的場(chǎng)景都應(yīng)該避免使用副作用
  • 副作用不能被濫用,因?yàn)槭褂玫恼_性和效率都無(wú)法保證,因?yàn)?strong>Stream會(huì)并行執(zhí)行
  • 大多數(shù)使用副作用的地方都可以使用歸約操作來(lái)更安全和有效地完成

// ======================== 錯(cuò)誤的收集方式 ========================
ArrayList<String> results = new ArrayList<>();
stream.filter(s -> pattern.matcher(s).matches()).forEach(s -> results.add(s));

// ======================== 正確的錯(cuò)誤收集方式 ====================
List<String> results = stream.filter(s -> pattern.matcher(s).matches()).collect(Collectors.toList());

• 根據(jù)不同的Stream結(jié)束操作,需要返回結(jié)果的流水線結(jié)果存儲(chǔ)在不同的位置:
  
  在這里插入圖片描述
• 對(duì)于表中返回boolean或者Optional(存放一個(gè)值的容器)的操作,由于返回一個(gè)值,只需要在對(duì)應(yīng)的Sink中記錄這個(gè)值,等到執(zhí)行結(jié)束時(shí)返回
• 對(duì)于規(guī)約操作,最終結(jié)果存放在用戶調(diào)用時(shí)指定的容器中,容器類(lèi)型通過(guò)收集器指定. collect(), reduce(), max(), min() 都是規(guī)約操作.盡管max() 和min() 也是返回一個(gè)Optional, 但事實(shí)上底層是通過(guò)reduce() 方法實(shí)現(xiàn)的
• 對(duì)于返回?cái)?shù)組的情況,結(jié)果放在數(shù)組中. 但是在最終返回?cái)?shù)組之前,結(jié)果存儲(chǔ)在Node的數(shù)據(jù)結(jié)構(gòu)中:
  • Node是一種多叉樹(shù)結(jié)構(gòu),元素存儲(chǔ)在樹(shù)的葉子中,并且一個(gè)葉子節(jié)點(diǎn)可以存放多個(gè)元素.這樣執(zhí)行起來(lái)方便

總結(jié)

• 能夠使用Lambda表達(dá)式的依據(jù)是必須有響應(yīng)的函數(shù)接口(內(nèi)部只有一個(gè)抽象方法的接口)
• Lambda表達(dá)式主要用來(lái)定義行內(nèi)執(zhí)行的方法類(lèi)型接口
• Lambda表達(dá)式免去了使用匿名方法和匿名內(nèi)部類(lèi)的麻煩,使Java有了簡(jiǎn)單強(qiáng)大的函數(shù)化編程能力