函数式编程（一） lambda、FunctionalInterface、Method Reference
函数式编程（二） Stream
Collector是Stream的一个重要部分，是一种汇聚操作，与函数式编程(二)中的reduce不同的是，Collector是一种可变汇聚，Collector只是一个接口，主要作为stream的<R, A> R collect(Collector<? super T, A, R> collector)方法的入参使用。Collector可用于将流中的元素汇聚到集合、使用StringBuilder连接String、计算一些统计信息等。汇聚操作可以串行或并行执行，Collectors提供了许多常用的可变汇聚实现。
JavaDoc: A mutable reduction operation that accumulates input elements into a mutable result container, optionally transforming the accumulated result into a final representation after all input elements have been processed. Reduction operations can be performed either sequentially or in parallel.
Examples of mutable reduction operations include: accumulating elements into a Collection; concatenating strings using a StringBuilder; computing summary information about elements such as sum, min, max, or average, etc. The class Collectors provides implementations of many common mutable reductions.
class Student{
String name;
int age;
int score;
public Student(String name, int age, int score) {
super();
this.name = name;
this.age = age;
this.score = score;
}
@Override
public String toString() {
return "Student [name=" + name + ", age=" + age + ", score=" + score + "]";
}
}
数据源：
List<Student> list = Arrays.asList(new Student("wang", 20, 90),
new Student("zhao", 30, 80), new Student("li", 25, 99),
new Student("sun", 20, 80), new Student("zhou", 30, 70));
首先给出一个按照学生年龄分组的例子：
//按照学生年龄分组，value为该年龄的学生的平均分数
Map<Integer,Double> map = list.stream().collect(
Collectors.groupingBy(Student::getAge, TreeMap::new, Collectors.averagingInt(Student::getScore)));
首先Collectors.groupingBy(,,**)返回的是实现了Collector接口的CollectorImpl类型的对象。其作为collect的参数进行汇聚，同时使用了Collectors的工厂方法groupingBy和averagingInt。
类型擦除与堆污染
在Collector的章节中加入类型擦除与堆污染的知识比较突兀，但如果不理解Java中的类型擦除与堆污染的相关知识去理解Collectors的源码是比较有难度的。Oracle官方Type Erasure & Heap Pollution 。关于泛型的处理，C++与Java存在很多明显的区别，最大的差异就是在C++中Foo<A>和Foo<B>会编译产生两个类文件，而在Java中Foo<A>与Foo<B>只会产生一个Foo类文件。Java为了对泛型进行限定，在泛型编程中也引入了extends和super关键字来限制泛型的区间。
1.类型擦除
public class Node<T> {
private T data;
private Node<T> next;
public Node(T data, Node<T> next) {
this.data = data;
this.next = next;
}
public T getData() { return data; }
}
由于类型参数T没有限制，Java编译器会将其替代为Object：
public class Node {
private Object data;
private Node next;
public Node(Object data, Node next) {
this.data = data;
this.next = next;
}
public Object getData() { return data; }
}
如果类型参数T加入类型限制:
public class Node<T extends Comparable<T>> {
private T data;
private Node<T> next;
public Node(T data, Node<T> next) {
this.data = data;
this.next = next;
}
public T getData() { return data; }
}
Java编译器会将被限制的类型参数T替换为约束类型：
public class Node {
private Comparable data;
private Node next;
public Node(Comparable data, Node next) {
this.data = data;
this.next = next;
}
public Comparable getData() { return data; }
}
2.堆污染
class HeapPollution {
public static <A> void setToMap(Function<Set<String>, Map<String, String>> f, Map<Integer, A> map) {
@SuppressWarnings("unchecked")
Function<A, A> func = (Function<A, A>) f;
System.out.println(map.get(1).getClass());//代码不严谨，只是为了打印出类型
map.replaceAll((k, v) -> func.apply(v));
System.out.println(map.get(1).getClass());//代码不严谨，只是为了打印出类型
}
public static void setToMapError(Function<Set<String>, Map<String, String>> f, Map<Integer, Set<String>> map) {
Function func = f;
Function<Set<String>, Set<String>> ff = (Function<Set<String>, Set<String>> )func;
map.replaceAll((k, v) -> ff.apply(v));
}
public static void main(String[] args) {
Map<Integer, Set<String>> map = new HashMap<>();
map.put(1, new HashSet<>(Arrays.asList("a", "b", "c")));
map.put(2, new HashSet<>(Arrays.asList("d", "e", "f")));
Function<Set<String>, Map<String, String>> finisher = set -> {
Map<String, String> map1 = new HashMap<>();
for (String ss : set) {
map1.put(ss, ss);
}
return map1;
};
setToMap(finisher, map);
// setToMapError(finisher, map);
}
}
代码中定义Map<Integer, Set<String>>类型的map变量，使用泛型setToMap可以将map的类型转化为Map<Integer, Map<String，String>>，这就是堆污染的含义。代码中也定义了不适用泛型进行转化的代码setToMapError，虽然编译可以通过，但是会出现运行时的ClassCastException。
public static <A> void setToMap(Function<Set<String>, Map<String, String>> f, Map<Integer, A> map) {
map.replaceAll((k, v) -> {
return f.apply(v);
});
}
public static void setToMapError(Function<Set<String>, Map<String, String>> f, Map<Integer, Set<String>> map) {
map.replaceAll((k, v) -> {
return (Set) f.apply(v);
});
}
对比setToMap与setToMapError的编译的字节文件的差异，只是在setToMapError中多了一个Set类型的强制类型转化。而这正是setToMap类型擦除所致。
Heap Pollution虽然有时会带来灾难，但巧妙利用它也可更方便的实现功能，而Collectors的代码中几处恰恰使用了Type Erasure引起的Heap Pollution实现需求(如Collectors.groupingBy)。
Collector
对于Collector<T, A, R>接口，T：流元素的类型，即执行汇聚操作的元素类型；A： 可变的汇聚类型，为中间结果类型，通常作为实现细节隐藏，可以是一个容器(如Collectors.toList())，可以是StringBuilder(如Collectors.joining())，可以是一个对象，但不会是简单数据类型(如int)，原因是可变汇聚，要对流的元素类型为T的每个元素和A执行汇聚操作，简单类型不会体现汇聚结果，简单类型一般用于reduce不变汇聚中使用，可以参照函数式编程(二)中reduce与collect的等价代码体会。R是汇聚操作的结果类型。了解三个类型的含义对于理解Collector接口的方法及后续Collectors的实现大有裨益。
Stream的collect的下述重载方法对于理解Collector接口非常重要。
<R> R collect(Supplier<R> supplier, BiConsumer<R, ? super T> accumulator,BiConsumer<R, R> combiner);
//等价代码
R result = supplier.get();
for (T element : this stream)
accumulator.accept(result, element);
return result;
1.Supplier<A> supplier()
supplier的作用是初始化一个A类型的结果容器。注意:返回的是中间结果类型，即A类型，而非结果类型R。查看源码Collectors.toList()的supplier为ArrayList::new，Collectors.joining()为StringBuilder::new。
2.BiConsumer<A, T> accumulator()
accumulator是累积器，对 supplier()返回的结果容器和流中的每个元素执行accept方法，即将流中的每个元素折叠到结果容器里。查看源码Collectors.toList()的accumulator为List::add(方法引用 --- 类名::实例方法名)，Collectors.joining()为StringBuilder::append。
3.BinaryOperator<A> combiner()
combiner是用于两个并行结果的合并，通常是将一个并行结果合并到另一个并行结果中。Collectors.toList()的combiner为一个lambda表达式(left, right) -> { left.addAll(right); return left; }，作用是两个并行结果的合并。Collectors.joining()的combiner为(r1, r2) -> { r1.append(r2); return r1; }。结合代码理解fold的含义。
4.Function<A, R> finisher()
finisher的主要作用是完成中间结果类型A到R类型的转化。如果包含IDENTITY_FINISH属性，则直接将A强制类型转化为R。Collectors.toList()使用A到R的强制类型转化(都是List类型)，而Collectors.joining()为StringBuilder::toString，实现了中间结果类型StringBuilder(A)到结果类型String(R)的转化。
5.Set<Characteristics> characteristics()
characteristics方法用来表征Collector的特性。
| 特性 | 释义 |
|---|---|
| CONCURRENT | 表示结果容器A支持并发，如果指定该特征，则在并行计算时，多个线程对同一个结果进行汇聚(supplier只调用一次，combiner不会被调用)；如果不指定，则在并行计算时，多个线程对不同的结果进行汇聚(supplier调用多次，combiner进行并行结果的合并) |
| UNORDERED | 表示汇聚操作不需要保留流中元素的顺序，当结果容器无明显的顺序时设置 |
| IDENTITY_FINISH | 当A类型与R类型相同时设置，此时finisher不执行，直接进行A到R的强制类型转换 |
Collectors
Collectors：Implementations of Collector that implement various useful reduction operations, such as accumulating elements into collections, summarizing elements according to various criteria, etc.
| Set | 汇聚操作包含特性 |
|---|---|
| CH_CONCURRENT_ID | CONCURRENT、UNORDERED、IDENTITY_FINISH |
| CH_CONCURRENT_NOID | CONCURRENT、UNORDERED |
| CH_ID | IDENTITY_FINISH |
| CH_UNORDERED_ID | UNORDERED、IDENTITY_FINISH |
| CH_NOID | 空 |
1.Collectors源码分析
Collectors为提供Collector的工厂方法，代码虽短，但由于大量的模板类型参数，如果含义不清楚，则理解将非常困难。
-
Collector<T, A, R>
T：流元素的类型，即执行汇聚操作的元素类型;
A：可变的汇聚类型，为中间结果类型;
R：是汇聚操作的结果类型 -
CollectorImpl
Collectors的内部类CollectorImpl是Collector的实现接口，有两个构造函数，只是将Collector的5个接口方法实现传入进行构造，差异是如果Collector的特征包含IDENTITY_FINISH，则直接强制类型转化即可，无需调用finisher。 -
groupingBy源码分析
groupingBy.png
代码重要部分都进行了标注，groupingBy收集器无非是通过包装downstream的5个Collector方法来形成自身的5个对外接口方法。代码使用了类型擦除进行强制类型转化，使用堆污染进行数据转换。
写一个收集器MyCollector 作为groupingBy的下级收集器downstream。
class MyCollector implements Collector<Student, Set<String>, Map<Integer, List<String>>> {
@Override
public Supplier<Set<String>> supplier() {
return HashSet::new;
}
@Override
public BiConsumer<Set<String>, Student> accumulator() {
return (set, s) -> set.add(s.getName());
}
@Override
public BinaryOperator<Set<String>> combiner() {
return (set1, set2) -> {
set1.addAll(set2);
return set1;
};
}
@Override
public Function<Set<String>, Map<Integer, List<String>>> finisher() {
return set -> {
Map<Integer, List<String>> map = new HashMap<>();
for (String str : set) {
List<String> list = map.getOrDefault(str.length(), new ArrayList<>());
list.add(str);
}
return map;
};
}
@Override
public Set<Characteristics> characteristics() {
return Collections.emptySet();
}
}
public class App {
public static void main(String[] args) {
List<Student> list = Arrays.asList(new Student("wang", 20, 90), new Student("zhao", 30, 80),
new Student("li", 25, 99), new Student("sun", 20, 80), new Student("zhou", 30, 70));
MyCollector mc = new MyCollector();
HashMap<Integer, Map<Integer, List<String>>> map =
list.stream().collect(Collectors.groupingBy(Student::getAge, HashMap::new, mc));
System.out.println(map);
}
}
对于示例代码，T为Student，K为学生年龄Integer类型，D为MyCollector的结果类型即Map<Integer, List<String>>，该map的键是学生名字的长度，值是学生名字列表。A为MyCollector的中间类型即Set<String>(学生名字列表)，M为Map<K, D>，即Map<Integer, Map<Integer, List<String>>>。
-
partitioningBy源码分析
partitioningBy.png
源码引入Partition类型进行分区，在该方法中都是new的对象不会涉及Heap Pollution，只要理解Partition的类型在执行finisher之前是Map<Boolean, A>，在执行finisher之后为Map<Boolean, D>，supplier是Map<Boolean, A>的工厂方法，代码自然比较清楚，不再冗述。
2.Collectors接口使用
-toCollection
public static <T, C extends Collection<T>>
Collector<T, ?, C> toCollection(Supplier<C> collectionFactory) {
return new CollectorImpl<>(collectionFactory, Collection<T>::add,
(r1, r2) -> { r1.addAll(r2); return r1; },
CH_ID);
}
源码比较简单，此收集器是将流中的元素收集到一个Collection中。
//将学生名字收集到LinkedList集合中
LinkedList<String> linkedList =
list.stream().map(Student::getName).collect(Collectors.toCollection(LinkedList::new));
-toList
public static <T>
Collector<T, ?, List<T>> toList() {
return new CollectorImpl<>((Supplier<List<T>>) ArrayList::new, List::add,
(left, right) -> { left.addAll(right); return left; },
CH_ID);
}
此收集器是将流中的元素收集到一个ArrayList中。
-toSet
public static <T>
Collector<T, ?, Set<T>> toSet() {
return new CollectorImpl<>((Supplier<Set<T>>) HashSet::new, Set::add,
(left, right) -> { left.addAll(right); return left; },
CH_UNORDERED_ID);
}
此收集器是将流中的元素收集到一个HashSet中。
-joining(简单拼接，无前缀、后缀、分隔符)
public static Collector<CharSequence, ?, String> joining() {
return new CollectorImpl<CharSequence, StringBuilder, String>(
StringBuilder::new, StringBuilder::append,
(r1, r2) -> { r1.append(r2); return r1; },
StringBuilder::toString, CH_NOID);
}
此收集器是将流中的元素(CharSequence类型)收集到一个String中。此收集器使用StringBuilder进行收集。
-joining(含前缀、分隔符、后缀)
public static Collector<CharSequence, ?, String> joining(CharSequence delimiter,
CharSequence prefix,
CharSequence suffix) {
return new CollectorImpl<>(
() -> new StringJoiner(delimiter, prefix, suffix),
StringJoiner::add, StringJoiner::merge,
StringJoiner::toString, CH_NOID);
}
此收集器是将流中的元素(CharSequence类型)收集到一个String中。此收集器使用StringJoiner进行收集。
//将所有学生姓名进行拼接，加前缀<name>，后缀</name>，分割符,
String name = list.stream().map(Student::getName).
collect(Collectors.joining(",", "<name>", "</name>"));
-mapping
public static <T, U, A, R>
Collector<T, ?, R> mapping(Function<? super T, ? extends U> mapper,
Collector<? super U, A, R> downstream) {
BiConsumer<A, ? super U> downstreamAccumulator = downstream.accumulator();
return new CollectorImpl<>(downstream.supplier(),
(r, t) -> downstreamAccumulator.accept(r, mapper.apply(t)),
downstream.combiner(), downstream.finisher(),
downstream.characteristics());
}
此收集器是首先将流中的元素利用mapper进行从T类型到U类型的转化，然后再使用downstream进行收集。其与Stream.map(mapper).collect(downstream)的效果相同，但更多用于多级收集中。The mapping() collectors are most useful when used in a multi-level reduction, such as downstream of a groupingBy orpartitioningBy.
//将所有学生姓名进行拼接，加前缀<name>，后缀</name>，分割符","，使用mapping
String name = list.stream().collect
(Collectors.mapping(Student::getName, Collectors.joining(",", "<name>", "</name>")));
-collectingAndThen
public static<T,A,R,RR> Collector<T,A,RR> collectingAndThen(Collector<T,A,R> downstream,
Function<R,RR> finisher) {
Set<Collector.Characteristics> characteristics = downstream.characteristics();
if (characteristics.contains(Collector.Characteristics.IDENTITY_FINISH)) {
if (characteristics.size() == 1)
characteristics = Collectors.CH_NOID;
else {
characteristics = EnumSet.copyOf(characteristics);
characteristics.remove(Collector.Characteristics.IDENTITY_FINISH);
characteristics = Collections.unmodifiableSet(characteristics);
}
}
return new CollectorImpl<>(downstream.supplier(),
downstream.accumulator(),
downstream.combiner(),
downstream.finisher().andThen(finisher),
characteristics);
}
作用是：将downstream的结果利用finisher完成从R类型到RR类型的转化。
//利用collectingAndThen将ArrayList的结果类型转化为HashSet
HashSet<Student> set = list.stream().collect
(Collectors.collectingAndThen(Collectors.toList(), HashSet::new));
-groupingBy
源码前面已分析过，该收集器是完成分组，并且返回的map类型不是并发的。Returns a Collector implementing a cascaded "group by" operation on input elements of type T, grouping elements according to a classification function, and then performing a reduction operation on the values associated with a given key using the specified downstream Collector. The Map produced by the Collector is created with the supplied factory function.
//将学生按照年龄进行分组
TreeMap<Integer, Set<Student>> map = list.stream().collect
(Collectors.groupingBy(Student::getAge, TreeMap::new, Collectors.toSet()));
-groupingByConcurrent
该收集器也是完成分组，与groupingBy不同，其返回的map是并发的，源码与groupingBy函数基本类似，只是考虑了downstream是否存在CONCURRENT特性，如不存在，则加synchronized进行并发的同步。
-partitioningBy
该收集器是完成分区，返回的类型为Map<Boolean, ?>，源码中没有提供对分区的并发处理。为什么呢？只有两组数据，没有必要。返回的Map不支持并发。
//按照年龄是否大于25进行分组
Map<Boolean, List<Student>> map = list.stream().collect
(Collectors.partitioningBy(s ->s.getAge() > 25));
关于Collector、Collectors的内容阐述完毕，后续对stream的源码继续分析。
WalkeR_ZG
