Java 8中的Lambda表达式详细介绍

Java 8中的Lambda表达式

• • 
    
• ​​Lambda表达式​​
• ​​变量作用域​​
• ​​使用示例​​

• • 
    
• ​​匿名内部类​​
• ​​带参函数​​
• ​​Collection​​

• • 
    
• ​​forEach​​
• ​​removeIf​​
• ​​replaceAll​​
• ​​sort​​
• ​​spliterator​​
• ​​stream和parallStream​​
  • 
    

• ​​Map​​

• • 
    
• ​​forEach​​
• ​​getOrDefault​​
• ​​putIfAbsent​​
• ​​remove​​
• ​​replace​​
• ​​replaceAll​​
• ​​merge​​
• ​​compute​​
• ​​computeIfAbsent​​
• ​​computeIfPresent​​
  • 
    

• ​​Streams API​​

• • 
    
• ​​forEach​​
• ​​filter​​
• ​​distinct​​
• ​​sorted​​
• ​​map​​
• ​​flatMap​​
  • 
    

• ​​Stream API 高级​​

• • 
    
• ​​reduce​​
• ​​collect​​
• ​​接口的静态方法和默认方法​​
• ​​方法引用​​
• ​​Collector​​
• ​​使用collect()生成Collection​​
• ​​使用collect()生成Map​​
• ​​使用collect()做字符串join​​
  • 
    

• ​​Stream Pipelines​​

• • 
    
• ​​Stream Pipeline实现方案​​
• ​​Stream Pipeline解决方法​​

• • 
    
• ​​操作如何记录?​​
• ​​操作如何叠加?​​
• ​​叠加操作如何执行?​​
• ​​执行后的结果位置​​
  • 
    

• ​​总结​​
  • 
    

Lambda表达式

• • 
    
• ​JVM​内部是通过​invokedynamic​指令来实现​Lambda​表达式的
• ​Lambda​中允许将一个函数作为方法的参数,即函数作为参数传递进方法中
• 使用​Lambda​表达式可以使代码更加简洁
  • 
    

变量作用域

• • 
    
• ​Lambda​表达式只能引用标记了​final​的外层局部变量.即不能在​Lambda​表达式内部修改定义在作用域外的局部变量,否则会导致报错
• ​Lambda​表达式中可以直接访问外层的局部变量
• ​Lambda​表达式中外层局部变量可以不用声明为​final,​ 但是必须不可被后面的代码修改,即隐性地具有final的语义
• ​Lambda​表达式中不允许声明一个与外层局部变量同名的参数或者局部变量
  • 
    

使用示例

匿名内部类

• ​匿名内部类:​ 匿名内部类仍然是一个类,不需要指定类名,编译器会自动为该类取名
• ​​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();; } } ​​

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

• ​使用Lambda表达式实现匿名内部类:​

带参函数

• ​带参函数的简写:​

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();       
  }
});

• 上述代码通过内部类重载了​Comparator​接口的​compare()​ 方法来实现比较逻辑. 采用​Lambda​表达式可简写如下:

List<String> list = Arrays.asList("I", "love", "you", "too");
Collections.sort(list, (s1, s2) -> { // 省略参数表类型
  if (s1 == null)
    return -1;
  if (s2 == null)
    return 1;
  return s1.length() - s2.length();
});

• • 
    
• 上述代码根内部类的作用一样
• 除了省略了接口名和方法名,代码中的参数类型也可以省略
• 因为​javac​的​类型推断​机制,编译器能够根据上下文信息推断出参数的类型
  • 
    

Collection

forEach

• 增强型for循环:

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

• 使用​forEach()​ 方法结合匿名内部类实现:

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表达式实现如下:

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

上述代码给​forEach()​ 方法传入一个​Lambda​表达式,不需要知道​accept()​ 方法,也不需要知道​Consumer​接口,类型推导已经完成了这些

removeIf

• • 
    
• 该方法签名: ​boolean removeIf(Predicate<? super E> filter);​

• ​删除容器中所有满足filter指定条件的元素​

• ​Predicate​是一个函数接口,里面有一个待实现的方法boolean test(T t)

• 如果需要在迭代过程中对容器进行删除操作必须使用​迭代器,​ 否则会抛出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()​ 方法结合匿名内部类实现:

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结合Lambda表达式实现:

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

使用​Lambda​表达式不需要记忆​Predicate​接口名,也不需要记忆​test()​ 方法名,只需要此处需要一个返回布尔类型的​Lambda​表达式

replaceAll

• • 
    
• 该方法签名: ​void replaceAll(UnaryOperator operator);​

• ​对每个元素执行operator指定的操作,并用操作结果来替换原来的元素​

• ​UnaryOperator​是一个函数接口,里面有待实现的方法​T apply(T t)​

• 使用下标实现元素替换:
  • 
    

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​结合匿名内部类实现:

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;
  }
});

代码调用​replaceAll()​ 方法,并使用匿名内部类实现​UnaryOperator​接口

• 使用​Lambda​表达式实现:

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);​

• ​根据c指定的比较规则对容器进行排序​

• ​Comparator接口中需要实现接口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()​ 方法,结合​Lambda​表达式:

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

spliterator

• 该方法签名: ​Spliterator spliterator();​

• • 
    
• ​Spliterator​既可以像​Iterator​那样逐个迭代,也可以批量迭代,批量迭代可以降低迭代的开销
• ​Spliterator​是可拆分的,一个​Spliterator​可以通过调用​Spliterator trySplit()​ 方法来尝试分成两个.一个是​this,​ 一个是新返回的元素.这两个迭代器代表的元素没有重叠
• 可通过多次调用​Spliterator.trySplit()​ 方法来分解负载,以便于多线程处理
  • 
    

stream和parallStream

• • 
    
• ​Stream()​ 和​parallStream()​ 分别返回该容器的​Stream​视图表示
• ​parallStream()​ 返回并行的​Stream​
• ​Stream​是​Java函数式编程​的核心类
  • 
    

Map

forEach

• • 
    
• 该方法签名: ​void forEach(BiConsumer<? super K,? super V> action);​

• 对​Map​中的每个映射执行​action​操作

• ​BiConsumer​是一个函数接口,里面有一个待实现方法 ​void accept(T t, U u);​

• 使用​Java 7​之前的方式输出​Map​中所有的对应关系:
  • 
    

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()​ 方法,结合匿名内部类:

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​表达式:

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​中对应的​value,​ 如果没有找到则返回​defaultValue​

• 查询Map中指定键所对应的值,如果不存在则返回​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​时,才将​value​指定的值放入到​Map​中,否则不对​Map​做修改
• 该方法将判断和赋值合二为一,使用起来更加方便
  • 
    

remove

• • 
    
• 该方法签名: ​remove(Object key);​

• 根据指定的​key​值删除​Map​中映射关系

• 该方法签名: ​remove(Object key, Object value);​

• 只有在当前​Map​中​key​正好映射到​value​时才删除该映射

replace

• • 
    
• 该方法签名: ​replace(K key, V value);​

• 只有在当前​Map​中​key​的映射存在时才用​value​去替换原来的值

• 该方法签名: ​replace(K key, V oldValue, V newValue);​

• 只有在当前​Map​中​key​的映射存在且等于​oldValue​时,才用​newValue​去替换原来的值,否则不做任何操作

replaceAll

• • 
    
• 该方法签名: ​replaceAll(BiFunction<? super K, ? super V, ? extends V> function);​

• • 
    
• 对​Map​中的每个映射执行​function​操作,并用​function​的执行结果替换原来的​value​
• 其中​BiFunction​是一个函数接口,里面有一个待实现的方法​R apply(T t, U u)​
  • 
    

• 使用​Java 7​以前的方式将Map中的映射关系的单词都转换成大写:
  • 
    

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​方法结合匿名内部类:

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​表达式实现:

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​对应的映射不存在或者为​null,​ 则将​value,​ value不可能为null关联到​key​上
• 否则执行​remappingFunction,​ 如果执行结果​非null,​ 则用该结果与​key​关联,否则在​Map​中删除​key​的映射
• 其中​BiFunction​是一个函数接口,里面有一个待实现方法​R apply(T t, U u)​
  • 
    

• ​merge()方法语义复杂,但使用的方式明确,经典的使用场景: 将新的错误信息拼接到原来的信息上​:
  • 
    

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

compute

• • 
    
• 该方法签名: ​compute(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction);​

• 将​remappingFunction​计算的结果关联到​key​上,如果计算结果为​null,​ 则在​Map​中删除​key​的映射

• 使用​compute​实现将新的错误信息拼接到原来的信息上:
  • 
    

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

computeIfAbsent

• • 
    
• 该方法签名: ​V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction);​

• • 
    
• 只有在当前​Map​中不存在​key​值的映射或映射值为​null​时,才调用​mappingFunction,​ 并在​mappingFunction​执行结果非​null​时,将结果跟​key​关联
• ​Function​是一个函数接口,里面有待实现方法​R apply(T t)​
  • 
    

• ​computeIfAbsent()​ 常用来对​Map​的某个​key​值建立初始化映射.比如在实现一个多值映射时 ​,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​表达式实现:

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()​ 相反
• 只有当前​Map​中存在​key​值的映射且非​null​时,才调用​remappingFunction,​ 如果​remappingFunction​执行结果为​null,​ 则删除​key​的映射,否则使用该结果替换​key​原来的映射
  • 
    

• 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函数式编程主角
  • stream不是某种数据结构,只是一种数据源视图
  • 这里的数据源可以是:
  • 
    
    • • 
        
    • ​数组​
    • ​Java容器​
    • ​I/O channel​
      • 
        
  • 
    
    • 
      
• 
  
• ​stream​是一个数据源视图,需要调用对应的工具方法创建一个​stream:​
• 
  
  • • 
      
  • 调用​Collection.stream()​ 方法
  • 调用​Collection.parallelStream()​ 方法
  • 调用​Arrays.stream(T[] array)​ 方法
    • 
      
• 
  
• ​stream接口继承关系:
• 图中​4​种​stream​接口继承自​BaseStream​:
• 
  
  • • 
      
  • ​IntStream,LongStream,DoubleStream​对应三种基本类型​int, long, double.​ 不是对应相应的包装类型
  • ​Stream​对应所有剩余类型的​stream​视图
    • 
      
• 
  
• 为不同的数据类型设置不同的stream接口:
• 
  
  • • 
      
  • 提高性能
  • 增加特定接口函数
    • 
      
• 
  
• ​尽管stream是容器调用Collection.stream()方法得到的. stream和collections有以下不同点:​
• 
  
  • • 
      
  • ​无存储:​ ​stream​不是一种数据结构,只是数据源的一个视图.数据源可以是数组,​Java​容器或​I/O channel​等
  • ​函数式编程:​ 对​stream​的修改都不会修改背后的数据源:比如对​stream​执行过滤操作并不会删除被过滤的元素,而是会产生一个不包含过滤元素的新​stream​
  • ​惰式执行:​ stream上的操作不会立即执行,只有等到真正需要stream执行的结果时才会执行
  • ​可消费性:​ stream只能被消费一次,一旦遍历过就会失效,就像容器的迭代器那样,想要再次遍历必须重新生成
    • 
      
• 
  
• ​对stream的操作分为两类:​
• 
  
  • • 
      
  • ​中间操作:​ intermediate operations. ​中间操作总会惰式执行.​ 调用中间操作只会生成一个标记了该操作的新stream
  • ​结束操作:​ terminal operations. ​结束操作会触发实际运算.​ 计算发生时,会把中间积攒的操作以pipeline的方式执行,这样可以减少迭代操作.计算完成之后stream就会失效
    • 
      
• 
  
• ​Stream接口常用方法:​
• 
  
  • • 
      
  • ​中间操作:​
  • 
    
    • • 
        
    • ​concat()​
    • ​distinct()​
    • ​filter()​
    • ​flatMap()​
    • ​limit()​
    • ​map()​
    • ​peek()​
    • ​skip()​
    • ​sorted()​
    • ​parallel()​
    • ​sequential()​
    • ​unordered()​
      • 
        
  • 
    
  • ​结束操作:​
  • 
    
    • • 
        
    • ​allMatch()​
    • ​anyMatch()​
    • ​collect()​
    • ​count()​
    • ​findAny()​
    • ​findFirst()​
    • ​forEach()​
    • ​forEachOrdered()​
    • ​max()​
    • ​min()​
    • ​noneMatch()​
    • ​reduce()​
    • ​toArray()​
      • 
        
  • 
    
    • 
      
• 
  
• 区分中间操作和结束操作就是看方法的返回值:
• 
  
  • • 
      
  • 返回值为​stream​的大都是​中间操作​
  • 否则是结束操作
    • 
      
• 
  
• ​Stream方法使用:​
• 
  
  • ​stream​与函数接口关系非常紧密,没有函数接口​stream​就无法操作
  • 
    
    • • 
        
    • ​函数接口是指内部只有一个抽象方法的接口​
    • 函数接口出现的地方都可以使用Lambda表达式
      • 
        
  • 
    
• 
  
  • 
    

forEach

• 该方法签名: ​void forEach(Consumer<? super E> action);​

• 对容器中的每个元素执行action指定的操作,即对元素进行遍历

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

• 由于​forEach()​ 是结束方法,所以上述方法会立即执行,输出所有字符串

filter

• 该函数原型: ​Stream< T > filter(Predicate<? super T> predicate);​

• 返回一个只包含满足​predicate​条件元素的​stream​

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

• • 
    
• 输出长度等于3的字符串you和too
• 由于​filter()​ 是个​中间操作,​ 如果只调用​filter()​ 不会有实际计算,因此不会输出任何信息
  • 
    

distinct

• 该函数原型: ​Stream< T > distinct();​

• 返回一个去除重复元素之后的​Stream​

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

• 输出去掉一个too之后的其余字符串

sorted

• 排序函数有两个:

• • 
    
• 自然顺序排序: ​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));

• 输出按照长度升序排序后的字符串

map

• 该函数原型: ​< R > Stream< R > map(Function<? super T, ? extends R> mapper);​

• • 
    
• 返回一个对当前所有元素执行​mapper​之后的结果组成的​Stream​
• 就是对每个元素按照某种操作进行转换 ,转换前后​Stream​中元素的个数不会改变,但是元素的类型取决于转换之后的类型
  • 
    

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

• 输出原字符串的大写形式

flatMap

• 该函数原型: ​< R > Stream< R > flatMap(Function<? super T, ? extends Stream<? extends R>> mapper);​

• • 
    
• 对每一个元素执行​mapper​的指定操作,并用所有​mapper​返回的​Stream​中的元素作为一个新的​Stream​作为最终返回结果
• 就相当于将原​Stream​中的所有元素都"摊平"之后组成的新​Stream.​ 转换前后元素的个数和类型都可能会改变
  • 
    

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));

• 原来的​stream​中有两个元素,分别是两个​List< Integer >,​ 执行​flatMap()​ 之后,将每一个​List​都"摊平"成一个个数字,所以会产生一个由5个数字组成的​Stream.​ 所以最终将输出1~5这5个数字

Stream API 高级

• • 
    
• ​归约操作:​ reduction operation

• • 
    
• 又称作折叠操作fold
• ​通过某个连接动作将所有元素汇总成一个汇总结果的过程​
• 元素求和,求最大值最小值,求元素总个数,将所有元素转换成一个列表或集合,都属于归约操作
  • 
    

• Stream类库中两个通用的规约操作:

• • 
    
• ​reduce()​
• ​collect()​
  • 
    

• 也有一些为了简化书写而设计的专用归约操作 ​: sum(), max(), min(), count()​ 等
  • 
    

reduce

• • 
    
• ​实现从一组元素中生成一个值​
• sum(),max(),min(),count()等都是reduce操作,将这些单独设为函数是因为经常使用
• ​reduce()的方法定义有三种重写形式:​
• 
  
  • • 
      
  • ​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);​
  • 虽然函数的定义越来越长,但是语义不变.多的参数是为了指明初始值(identity), 或者是指定并行执行时多个部分结果的合并方式(combiner)
    • 
      
• 
  
• 从一组单词中找出最长的单词.这里"大"的含义就是"长":

/*
* 找出最长的单词
*/
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());

• • 
    
• 选出最长的单词love
• 其中​Optional​是只有一个值的容器,使用​Optiona​l可以避免​null​值
  • 
    

/*
 * 求单词长度之和
 */
 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);   // 部分和拼接,并行执行时会用到 (3)
// int lengthSun = stream.mapToInt(str -> str.length()).sum();
System.out.println(lengthSum);

• • 
    
• 上述代码 ​(2)​ 处的累加器:

• • 
    
• 字符串映射成长度
• 并和当前累加和相加
• 使用​reduce()​ 函数将这两步合二为一,更有助于提升性能
  • 
    

• 同样也可以使用​map()​ 和​sum()​ 组合也可以达到目的
  • 
    

collect

• • 
    
• ​reduce()​ 的优点的是生成一个值,但是如果想要从​Stream​中生成一个集合或者​Map​等复杂对象时,就要用到​collect()​
• ​示例:​
  • 
    

/* 
 * 将Stream转换成容器或者Map
 */
Stream<String> stream = Stream.of("I", "love", "you", "too");
List<String> list = stream.collect(Collectors.toList());
Set<String> set = stream.collect(Collectors.toSet());
// 将Stream转换成Map
Map<String, Integer> map = stream.collect(Collectors.toMap(Function.identity(), String::length));

• • 
    
• 上述分别将Stream转换成List,Set,Map
• 需要注意的有:

• • 
    
• ​Function.identity()​
• ​String::length​
• ​Collectors​
  • 
    

接口的静态方法和默认方法

• ​Function​是一个接口 ​,Function.identity()​ 含义有两个方面:

• • 
    
• Java 8允许在接口中加入具体方法. 接口中的具体方法有两种:

• • 
    
• ​static:​ 静态方法,identity()就是Function接口的一个静态方法
• ​default:​ 默认方法
  • 
    

• ​Function.identity():​ 返回一个输出和输入一样的Lambda表达式对象,等价于 ​t -> t​ 形式的Lambda表达式
  • 
    

在Java 7之前要想在定义好的接口中加入新的抽象方法是很困难甚至不可能的,因为会所有实现了该接口的类都要重新实现.Java 8中的default方法就是用来解决这个问题,直接在接口中实现新加入的方法,引进了default方法之后,可以继续加入static方法来避免专门的工具类

方法引用

• • 
    
• 形如​String::length​的语法格式叫作方法引用(method reference),这种语法用来替代某些特定形式的Lambda表达式
• ​如果Lambda表达式的全部内容就是调用一个已有的方法,就可以用方法引用来代替Lambda表达式​
• 方法引用可以分为四类:

• • 
    
• ​引用静态方法:​ Integer :: sum
• ​引用某个对象的方法:​ list :: add
• ​引用某个类的方法:​ String :: length
• ​引用构造方法:​ HashMap :: new
  • 
    

Collector

• • 
    
• 收集器​Collector​是为​Stream.collect​方法打造的工具类
• 将一个Stream转换成一个容器或者Map至少需要考虑两个方面:

• • 
    
• ​目标容器是什么:​ ArrayList,HashSet还是TreeMap
• ​新元素如何添加到目标容器中:​ List.add()还是Map.put()
• 如果是并行进行规约,还要使得collect()如何做到将多个部分结果合并成一个
  • 
    

• ​collect()​ 方法定义: ​< R > R collect(Supplier< R > supplier, BiConsumer<R, ? super T> accumulator, BiConsumer<R, R> combiner);​

• • 
    
• 三个参数一次对应着上面的三条分析
• 每次调用​collect()​ 都要传入这三个参数很麻烦,所以使用收集器​Collector​对三个参数进行简单的封装
• 所以​collect()​ 另一个方法定义: ​< R, A> R collect(Collecor<? super T, A, R> collector);​
  • 
    

• ​Collectors​工具类可以通过静态方法生成各种常用的​Collector​
• 这样,将Stream规约成List可以通过如下两种方式:
  • 
    

/*
 * 将Stream规约成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);

• • 
    
• 通常情况下不需要手动指定​collect()​ 的三个参数,而是调用​collect(Collector<? super T, A, R> collector)​ 方法,并且参数中的​Collector​对象大都是直接通过​Collectors​工具类获得
• ​实际传入的收集器的行为决定collect()的行为​
  • 
    

使用collect()生成Collection

• • 
    
• 通过​collect()​ 方法将​Stream​转换成容器的方法中将​Stream​转换成​List​和​Set​是最常见的操作
• 在Collectors工具类中已经提供了对应的收集器:
  • 
    

/*
 * 将Stream转换成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());

• • 
    
• 由于返回结果是接口类型,所以并不清楚类库实际选择的容器类型什么
• 有时需要人为指定容器的实际类型,这个需求可以通过​Collectors.toCollection(Supplier< C > collectionFactory)​ 完成:
  • 
    

/*
 * 使用toCollection指定规约容器的类型
 */
 ArrayList<String> arrayList = stream.collect(Collectors.toCollection(ArrayList :: new));
 HashSet<String> hashSet = stream.collect(Collectors.toCollection(HashSet :: new));

• 分别指定规约结果为​ArrayList​和​HashSet​

使用collect()生成Map

• • 
    
• Stream依赖某种数据源,数据源可以是数组,容器等,但不能是Map
• 但是可以从Stream生成Map,要做的是确定好Map的key和value分别代表什么,这个在于要想清楚到底要干什么
• ​通常在三种情况下collect()的结果会是Map:​

• • 
    
• 使用​Collectors.toMap()​ 生成的收集器: 用户需要指定如何生成​Map​的​key​和​value​
• 使用​Collectors.partitioningBy()​ 生成的收集器: 对元素进行二分区操作时用到
• 使用​Collectors.groupingBy()​ 生成的收集器: 对元素做​group​操作时用到
  • 
    

• ​使用toMap()生成的收集器:​

• • 
    
• 这个是和​Collectors.toCollection()​ 并列的方法
• ​示例:​ 将学生列表转换成由<学生, GPA>组成的Map
  • 
    

/*
 * 使用toMap()统计学生的GPA
 */
 Map<Student, Double> studentToGPA = student.stream().collect(Collectors.toMap(Function.identity(), // 如何生成key
                                        student -> computeGPA(student))); // 如何生成value

• ​使用partitioningBy()生成的收集器:​

• • 
    
• 适用于将​Stream​中的元素依据某个二值逻辑(​Boolean​ 满足,不满足)分成互补相交的两部分
• ​示例:​ 将学生分成成绩及格和不及格的两部分
  • 
    

/*
 * 将学生成绩分为及格不及格两部分
 */
 Map<Boolean, List<Student>> passingFailing = students.stream().collect(Collectors.partitioningBy(s -> s.getGrade() >= PASS_THRESHOLD));

• ​使用groupingBy()生成的收集器:​

• • 
    
• 这是比较灵活的一种,与​SQL​中的​group by​语句类似
• 这里的​groupingBy​也是按照某个属性对数据进行分组,属性相同的元素会被对应到​Map​的同一个​key​上
• ​示例:​ 将员工按照部门进行分组
  • 
    

/*
 * 将员工按照部门进行分组
 */
 Map<Department, List<Employee>> byDept = employees.stream().collect(Collectors.groupingBy(Employee :: getDepartment));

• 有时候仅仅分组是无法满足要求的.在SQL中使用group by是为了方便更高级的查询:

• • 
    
• 先将员工按照部门分组
• 然后统计每个部门员工的人数
• ​增强版的groupingBy()能够满足这种需求:​

• • 
    
• 增强版的**groupingBy()**允许先对元素分组之后再执行某种运算,比如求和,计数,平均值,类型转换等
• 这种​先将元素分组的收集器​叫作​上游收集器​
• 然后​执行分组后的运算的收集器​叫作​下游收集器​
  • 
    

/*
 * 使用下游收集器统计每个部门的人数
 */
 Map<Department, Integer> totalByDept = employees.stream()
                         .collect(Collectors.groupingBy(Employee :: getDepartment,
                                        Collectors.counting()));

这个​groupingBy​和​SQL​相似,也是高度非结构化

• ​下游收集器还可以包含更下游的收集器:​

• • 
    
• 将员工按照部门分组
• 得到每个员工的名字字符串,而不是一个个Employee对象
  • 
    

/*
 * 按照部门对员工进行分组,并且只保留员工的名字
 */
 Map<Department, List<String>> byDept = employees.stream()
                         .collect(Collectors.groupingBy(Employee :: getDepartment,
                          Collectors.mapping(Employee :: getName,
                                  Collectors.toList())));

使用collect()做字符串join

• • 
    
• 字符串拼接时使用​Collectors.joining()​ 生成的收集器,代替for循环拼接
• ​Collectors.joining()​ 方法有三种重写形式,分别对应三种不同的拼接方式:
  • 
    

/*
 * 使用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工具类已经封装好的收集器,还可以自定义收集器.或者直接调用​collect(Supplier< R > supplier, BiConsumer<R, ? super T> accumulator, BiConsumer<R, R> combiner)​ 方法,收集需要的任何形式的信息

Stream Pipelines

• • 
    
• 通过使用Stream API中引起的疑问:

• • 
    
• ​如此强大的Stream API是如何实现的?​
• ​Pipeline是怎么执行的,每次调用都会迭代一次吗?​
• ​自动并行又是怎么做到的,线程个数是多少?​
  • 
    

• ​容器执行Lambda表达式的方式 - 以ArrayList.forEach()方法为例:​
  • 
    

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

• • 
    
• ​ArrayList.forEach()​ 方法的主要逻辑就是一个​for​循环,在该​for​循环里不断调用​action.accept()​ 回调方法完成对元素的遍历
• 回调方法在Java GUI的监听器中广泛使用,Lambda表达式的作用就是相当于一个回调方法
• ​Stream API​中大量使用​Lambda​表达式作为回调方法. 但想要理解​Stream,​ 关键的是:

• • 
    
• ​流水线​
• ​自动并行​
  • 
    

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

• • 
    
• 上述代码用来求出以字母 ​“A”​ 开头的字符串的最大长度:

• • 
    
• 一种直白的方式就是为每一次函数调用都执行一次迭代. 尽管这样做能够实现功能,但效率上是无法接受的
• 类库的实现是使用​Stream Pipeline​的方式巧妙地避免了多次迭代.基本思想就是​在一次迭代中尽可能多的执行用户指定的操作​
  • 
    

• ​Stream中的相关操作:​

• • 
    
• ​中间操作:​ Intermediate operations

• • 
    
• ​无状态:​ Stateless

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

• ​有状态:​ Stateful

• • 
    
• ​distinct()​
• ​sorted()​
• ​limit()​
• ​skip()​
  • 
    

• ​结束操作:​ Terminal operations

• • 
    
• ​短路操作:​ short-circuiting

• • 
    
• ​anyMatch()​
• ​allMatch()​
• ​noneMatch()​
• ​findFirst()​
• ​findAny()​
  • 
    

• ​非短路操作:​

• • 
    
• ​forEach()​
• ​forEachOrdered()​
• ​toArray()​
• ​reduce()​
• ​collect()​
• ​max()​
• ​min()​
• ​count()​
  • 
    

• ​Stream上的所有操作分为两类:​ 因为Stream底层对每一种情况的处理方式不同,所以要进行精细的划分

• • 
    
• ​中间操作:​ 中间操作只是一种标记

• • 
    
• ​无状态:​ 指元素的处理不受前面元素的影响,处理完一个元素就能立即知道结果
• ​有状态:​ 指元素的处理受到别的元素的影响,必须等到所有元素处理之后才能知道结果
  • 
    

• ​结束操作:​ 只有结束操作才会触发实际的计算

• • 
    
• ​短路操作:​ 指不用处理全部元素就可以返回结果
• ​非短路操作:​ 指对所有的元素处理后才可以返回结果
  • 
    

Stream Pipeline实现方案

• • 
    
• 一种直白的Stream Pipeline实现方案:
  
• ​求最长字符串的长度:​
• 
  
  • • 
      
  • 一种直白的实现方式是为每一次函数调用都执行一次迭代,并将处理中间结果发明放到某种数据结构中,比如数组,容器等
  • 
    
    • • 
        
    • 就是调用​filter()​ 方法后立即执行
    • 选出所有以​A​开头的字符串并放到一个列表​list1​中
    • 然后让​list1​传递给​mapToInt()​ 方法并立即执行
    • 生成的结果放到​list2​中
    • 最后遍历​list2,​ 找出最大的数字作为最终的结果
      • 
        
  • 
    
  • 这种实现方法实现简单直观,但存在两个明显的缺陷:
  • 
    
    • • 
        
    • ​迭代次数多:​ 迭代次数和函数的调用次数相等
    • ​频繁产生中间结果:​ 每次函数调用都产生一次中间结果,存储开销大
      • 
        
  • 
    
    • 
      
• 
  
• ​不使用Stream API在一次迭代中实现求最长字符串长度的方式:​
  • 
    

int longest = 0;
for (String str : strings) {
  if (str.startsWith("A")) {  // 类似filter(),保留以A开头的字符串
    int len = str.length(); // 类似mapToInt(),得到字符串的长度
    longest = Math.max( );
  }
}

• 采用这种方法不但减少了迭代次数,也避免了存储中间结果,这就是Stream Pipeline.将三个操作放在了一次迭代中

• 只要事先知道意图,总是能够采取上述方式实现与Stream API等价的功能

Stream Pipeline解决方法

• • 
    
• 由于Stream类库的设计者不知道用户意图,所以如何在无法假设用户行为的前提下,是类库的设计者要考虑的问题?
• 关于这个解决方法,可以​采用某种方式记录用户每一步的操作,当用户调用结束操作时将之前记录的操作叠加到一起在一次迭代中全部执行完成.​ 关于这种解决方法,需要解决以下四个问题:

• • 
    
• ​用户的操作如何记录?​
• ​操作如何叠加?​
• ​叠加之后的操作如何执行?​
• ​执行后的结果在哪里?​
  • 
    

操作如何记录?

• • 
    
• 这里的操作指的是​Stream中间操作​
• 很多Stream的操作会需要一个回调函数 - Lambda表达式,因此一个完整的操作应该是一个三元数组:
• 
  
  • ​<数据来源, 操作, 回调函数>​
• 
  
• ​Stream​中使用​Stage​的概念来描述一个完整的操作,并用某种实例化后的​PipelineHelper​来代表​Stage,​ 将具有先后顺序的各个​Stage​连到一起,就构成了整个​Stream Pipeline​
• ​Stream相关类和接口的继承关系图:
• ​IntPipeline.LongPipeline,DoublePipeline​三个类是专门为三种基本类型而不是包装类型定制的,与​ReferencePipeline​是并列关系
• 图中​Head​用于表示第一个​Stage,​ 即调用诸如​Collection.stream()​ 方法产生的​Stage,​ 很显然这个​Stage​中不包含任何操作
• ​StatelessOp​和​StatefulOp​分别表示无状态和有状态的​Stage,​ 对应于无状态和有状态的操作
• ​Stream Pipeline组织结构示意图:
• 通过​Collection.stream()​ 方法得到​Head,​ 即​Stage0,​ 紧接着调用一系列中间操作,不断产生新的​Stream​
• ​这些Stream对象以双向链表的形式组织在一起,构成整个流水线,由于每个Stage都记录了前一个Stage和本次操作的回调函数,依靠这种数据结构就能建立起对所有数据源的操作.​ 这就是Stream记录操作的方式
  • 
    

操作如何叠加?

• • 
    
• 通过上面的方法解决了操作记录问题,要想让Stream Pipeline起到应有的作用需要一种将所有操作叠加到一起的方案
• 因为只有当前​Stage​本身才知道该如何执行自己包含的操作.前面的​Stage​并不知道后面的​Stage​到底执行了哪种操作,以及回调函数是哪种形式,所以不能够从​Stream Pipeline​的​Head​开始依次执行每一步操作与回调函数来实现操作叠加
• 为了解决以上问题,就​需要某种协议来协调相邻Stage之间的调用关系,​ 这种协议由​Sink​接口完成 ​,Sink​接口包含的方法如下所示:
  • 
    

• • 
    
• 通过​Sink​协议,可以方便地进行相邻的​Stage​调用,每个​Stage​都会将自己的操作封装到一个​Sink​里,前一个​Stage​只需要调用最后一个​Stage​的​accept()​ 方法即可,并不需要知道Stage的内部是如何处理的
• 对于​有状态的操作,Sink​的​begin()​ 和​end()​ 方法是必须实现的:

• • 
    
• 比如​Stream.sorted()​ 是一个有状态的中间操作
• 对应的​Sink.begin()​ 方法可能创建一个存放结果的容器
• 而​accept()​ 方法负责将元素添加到容器中
• 最后​end()​ 负责对容器进行排序
  • 
    

• 对于​短路操作,Sink.cancellationRequest()​ 是必须实现的:

• • 
    
• 比如​Stream.findFirst()​ 是一个短路操作
• 只要找到一个元素 ​,cancellationRequested()​ 就应该返回​true,​ 以便调用者尽快结束查找
  • 
    

• Sink的四个接口方法互相协作,共同完成计算任务. ​实际上Stream API内部实现的本质,就是如何重载Sink的这四个接口方法​
• 根据​Sink​对操作的包装,就解决了​Stage​之间的调用问题,执行时只需要从流水线的​head​开始对数据源依次调用每个​Stage​对应的​Sink.{begin(), accept(), cancellationRequested(), end()}​ 方法
• ​示例:​ 一种可能的Sink.accept()方法流程
  • 
    

void accept(U u) {
  1. 使用当前Sink包装的回调函数处理u
  2. 将处理结果传递给Pipeline下游的Sink
}

• • 
    
• Sink接口的方法都是按照 ​[处理 -> 转发]​ 的模型实现
• ​示例:​ ​Stream​的中间操作是如何将自身的操作包装成​Sink​以及​Sink​如何将结果转发给下一个​Sink​的,Stream.map()方法如下
  • 
    

/*
 * Stream.map(),调用该方法将产生一个新的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()方法返回由回调函数包装成的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) {
          // 使用当前Sink包装的回调函数mapper处理u
          R r = mapper.apply(u)
          // 将处理结果传递给流水线下游的Sink
          downstream.accept(r);
        }
      };
     }
  };
}

• • 
    
• 将回调函数​mapper​包装到一个​Sink​中:

• • 
    
• ​Stream.map()​ 是一个无状态的中间操作,所以​map()​ 方法返回了一个​StatelessOp​内部类对象,一个新的​Stream​
• 调用这个新​Stream​的​opWripSink()​ 方法将得到一个包装了当前回调函数的​Sink​
  • 
    

• ​示例:​

• • 
    
• ​Stream.sorted()​ 方法将对​Stream​中的元素进行排序
• 这是一个有状态的中间操作,因为在读取所有元素之前是无法获得最终顺序的
• sorted()方法封装的Sink如下:
  • 
    

/*
 * Stream.sort()中的Sink实现
 */
 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) {
        ...
        // 创建一个存放排序元素的列表
        list = (size > 0) ? new ArrayList<T>((int)size) : new ArrayList<T>();
    }

    @Override
    public void end() {
        // 只有全部元素结束接收之后才能开始排序
        list.sort(comparator);
        downstream.begin(list.size());

        if (!cacellationWasRequested()) {
            // 如果下游Sink不包含短路操作,将处理结果传递给流水线下游的Sink
            list.forEach(downstream :: accept);
        } else {
            /*
             * 如果下游Sink包含短路操作:
             *  每次都调用cancellationRequested()询问是否可以结束处理
             */
             for (T t : list) {
                if (down.cancellationWasRequested()) {
                    break;
                }
                // 将处理结果传递给流水线下游的Sink
                downstream.accept();
             }
        }
        downstream.end();
        list = null;
    }

    @Override
    public void accept(T t) {
        /*
         * 使用当前Sink包装动作处理:
         *  将元素添加到中间列表中
         */
        list.add(t);
    }
 }

• ​Sink​中的四个接口方法的协作方式:

• • 
    
• 首先​begin()​ 方法获取参与排序的元素个数传递给​Sink.​ 方便确定中间结果容器的大小
• 然后通过​accept()​ 方法将元素添加到中间结果中,最终执行时调用者会不断调用该方法,直到遍历所有元素
• 最后​end()​ 方法返回给​Sink​所有元素遍历完毕,启动排序步骤,排序完成后将结果传递给下游的​Sink​
• 如果下游的​Sink​是短路操作,将结果传递给下游时不断询问下游​cancellationRequested()​ 是否可以结束处理
  • 
    

叠加操作如何执行?

• • 
    
• ​Sink​封装了​Stream​的每一步操作,并使用 ​[处理 -> 转发]​ 的模式来叠加操作.一旦调用某个​结束操作,​ 就会触发整个流水线的执行

• • 
    
• 结束操作之后不会有别的操作,所以结束操作不会创建新的流水线阶段​Stage.​ 流水线的链表不会再往后延伸
• 结束操作会创建一个包装了自己操作的​Sink,​ 这是最后一个​Sink,​ 不会有下游的​Sink.​ 所以这个​Sink​只需要处理数据而不需要将结果传递给下游的​Sink​
• 对于​Sink​的 ​[处理 -> 转发]​ 模型,结束操作的​Sink​就是调用链的出口
  • 
    

• ​上游Sink如何找到下游Sink:​

• • 
    
• 一种方案是在​PipelineHelper​中设置一个​Sink​字段,在流水线中找到下游​Stage​并访问​Sink​字段即可
• 在​Stream​中,设置了一个​SinkAbstractPipeline.opWrapSink(int flag, Sink downstream)​ 方法来得到​Sink.​ 该方法的作用:

• 返回一个新的包含了当前​Stage​代表的操作以及能够将结果传递给​downstream​的​Sink​对象

• ​使用一个新的Sink对象而不是返回一个Sink字段:​

• • 
    
• 因为使用​opWrapSink()​ 可以将当前操作与下游​Sink​的​downstream​参数结合成新的​Sink​
• 这样只要从流水线的最后一个​Stage​开始,不断调用上一个​Stage​的​opWrapSink()​ 方法直到最开始(不包括​stage0,​ 因为​stage0​代表数据源,不包含操作),就可以得到一个代表了流水线上所有操作的​Sink​
  • 
    

/**
 * AbstractPipeline.wrapStack():
 * 从下游向上游不断包装Sink,如果最初传入的Sink代表结束操作,函数返回时就可以得到一个代表了流水线上所有操作的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​上从开始到结束的所有操作都被包装到一个​Sink​里,执行这个​Sink​就相当于执行整个流水线:

/*
 * AbstractPipeline.copyInto():
 *  对spliterator代表的数据执行wrappedSink代表的操作
 */
 final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
  ...
  if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags)) {
    // 通知开始遍历履历
    wrappedSink.begin(spliterator.getExactSizeIfKnown());
    // 迭代
    spliterator.forEachRemaining(wrappedSink);
    // 通知遍历结束
    wrappedSink.end();
  }
  ...
 }

上述代码首先调用​wrappedSink.begin()​ 方法告诉​Sink​数据即将到来,然后调用​Spliterator​迭代器的​spliterator.forEachRemaining()​ 方法对数据进行迭代,最后调用​wrappedSink.end()​ 方法通知​Sink​数据处理结束

执行后的结果位置

• 首先不是所有的​Stream​结束操作都需要返回结果,有些操作只是为了使用副作用​Side-effects :​

• • 
    
• 比如​Stream.forEach()​ 方法将结果打印出来就是常见的副作用场景
• 事实上,除了打印之外的场景都应该避免使用副作用
• 副作用不能被滥用,因为使用的正确性和效率都无法保证,因为​Stream​会并行执行
• 大多数使用副作用的地方都可以使用​归约操作​来更安全和有效地完成
  • 
    

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

// ======================== 正确的错误收集方式 ====================
List<String> results = stream.filter(s -> pattern.matcher(s).matches()).collect(Collectors.toList());

• • 
    
• 根据不同的​Stream​结束操作,需要返回结果的流水线结果存储在不同的位置:
  
• 对于表中返回​boolean​或者​Optional​(存放一个值的容器)的操作,由于返回一个值,只需要在对应的​Sink​中记录这个值,等到执行结束时返回
• 对于规约操作,最终结果存放在用户调用时指定的容器中,容器类型通过收集器指定. ​collect(), reduce(), max(), min()​ 都是规约操作.尽管​max()​ 和​min()​ 也是返回一个​Optional,​ 但事实上底层是通过​reduce()​ 方法实现的
• 对于返回数组的情况,结果放在数组中. 但是在最终返回数组之前,结果存储在​Node​的数据结构中:
• 
  
  • ​Node​是一种多叉树结构,元素存储在树的叶子中,并且一个叶子节点可以存放多个元素.这样执行起来方便
• 
  
  • 
    

总结

• • 
    
• 能够使用​Lambda​表达式的依据是必须有响应的函数接口(内部只有一个抽象方法的接口)
• ​Lambda​表达式主要用来定义行内执行的方法类型接口
• ​Lambda​表达式免去了使用匿名方法和匿名内部类的麻烦,使​Java​有了简单强大的函数化编程能力
  •