Stream 中有两个操作比较繁琐，一个是reduce，另一个是collect
其中，collect的理解相对难一些，所以先研究collect，（汇聚操作）
0x00
1x00：Collector接口
<R, A> R collect(Collector<? super T, A, R> collector);
而collect中只有一个参数，那就是Collector对象，java.util.stream.Collector
这是一个接口，其功能是将流处理的结果，汇聚处理成最终的一个可变对象（容器）。
这个接口有5个方法：
-
Supplier<A> supplier();
创建一个结果容器 -
BiConsumer<A, T> accumulator();
将元素合并到结果容器中 -
BinaryOperator<A> combiner();
将两个结果容器合并 -
Function<A, R> finisher();
最终操作 -
Set<Characteristics> characteristics();
返回一个固定的特征集合
ps:讲述一下特征值
/**当前Collector支持并发，一般情况下UNORDERED也会一起放在characteristics中*/
CONCURRENT,
/** 当前Collector对操作元素是不会关心顺序的的 */
UNORDERED,
/**当前Collector没有Finisher*/
IDENTITY_FINISH
2x00：Collector接口方法详解
这些方法在处理时所起到的作用如流程图所示。
其中，串行操作会忽略combiner，每一个元素调用accumulator就可以完成
为了描述方便，后面描述基本以单个线程为主
同一性（identity）和结合性（associativity）保证了串行和并行执行结果的等价性
- 同一性
a = Combiner.apply(a, Supplier.get());
- 结合性
A a1 = supplier.get();
// 第一个参数是容器，第二个参数是集合的元素
accumulator.accept(a1, t1);
accumulator.accept(a1, t2);
R r1 = finisher.apply(a1); // result without splitting
A a2 = supplier.get();
accumulator.accept(a2, t1);
A a3 = supplier.get();
accumulator.accept(a3, t2);
R r2 = finisher.apply(combiner.apply(a2, a3));
如果：finisher.apply(a1).equals(finisher.apply(a2))==true;
则表示具有同一性，无序集合对同一性的要求会低一些。
collector可以代替reduce方法
3x00 Collectors类
Stream中的collect方法之一：
public final <R, A> R collect(Collector<? super P_OUT, A, R> collector) {
A container;
if (isParallel()
&& (collector.characteristics().contains(Collector.Characteristics.CONCURRENT))
&& (!isOrdered() || collector.characteristics().contains(Collector.Characteristics.UNORDERED))) {
container = collector.supplier().get();
BiConsumer<A, ? super P_OUT> accumulator = collector.accumulator();
forEach(u -> accumulator.accept(container, u));
}
else {
container = evaluate(ReduceOps.makeRef(collector));
}
return collector.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)
? (R) container
: collector.finisher().apply(container);
}
可以简单看出执行了collector中5个方法；
evaluate方法是对pipeline进行终止操作，从而产生一个结果。（源码比较深，感兴趣的自己可以深挖一波）；
Collectors一个常用的collector工厂，这个工厂就是collect中常用的收集方式的实现，建议扫一波源码
Collectors里的实现都是利用CollectorImpl实现的
这里挑一些说明一下
首先介绍一下一下Collector的常用的特性集合
这里贴出了大多Collector的实现所具有的特性。
- 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);
}
这里CH_ID是IDENTITY_FINISH，所以这个collector不执行finisher方法。
这个实现很明显，详细说明一下
supplier();创建一个结果容器ArrayList::newBiConsumer<A, T> accumulator();将元素合并到结果容器中List::addBinaryOperator<A> combiner();将两个结果容器合并(left, right) -> { left.addAll(right); return left; }
。
- toCollection(Supplier<C> collectionFactory)
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);
}
CH_ID,一样不执行finisher方法
流程与toList一样，不再作图
- reducing()
public static <T, U>
Collector<T, ?, U> reducing(U identity,
Function<? super T, ? extends U> mapper,
BinaryOperator<U> op) {
return new CollectorImpl<>(
boxSupplier(identity),
(a, t) -> { a[0] = op.apply(a[0], mapper.apply(t)); },
(a, b) -> { a[0] = op.apply(a[0], b[0]); return a; },
a -> a[0], CH_NOID);
}
CH_NOID，是个空Set，表示不具备任何特性。
supplier();创建一个结果容器boxSupplier(identity)BiConsumer<A, T> accumulator();将元素合并到结果容器中(a, t) -> { a[0] = op.apply(a[0], mapper.apply(t)); }BinaryOperator<A> combiner();将两个结果容器合并(a, b) -> { a[0] = op.apply(a[0], b[0]); return a; }Function<A, R> finisher();结束时数据整合a -> a[0]
这里使用数组的原因是为了兼容流元素为基本类型时不可更改，引用类型更方便的对数据进行操作。
- counting()
简单粗暴把reducing再包装了一下，作用是计算条目数，对嵌套集合有奇效。一般的集合用size就好了。嵌套的集合，几个flatMap展开之后，即可统计数量。很方便
public static <T> Collector<T, ?, Long> counting() {
return reducing(0L, e -> 1L, Long::sum);
}
看懂了reduce，就明白了
- minBy() 、maxBy()、summingInt()
// 传入一个元素比较器，获取最小的元素
public static <T> Collector<T, ?, Optional<T>> minBy(Comparator<? super T> comparator) {
return reducing(BinaryOperator.minBy(comparator));
}
// 传入一个比较器，获取最大的元素
public static <T> Collector<T, ?, Optional<T>> maxBy(Comparator<? super T> comparator) {
return reducing(BinaryOperator.maxBy(comparator));
}
// 统计元素中某个Int属性累加起来的总和，没有元素的时候为0
public static <T> Collector<T, ?, Integer> summingInt(ToIntFunction<? super T> mapper) {
return new CollectorImpl<>(
() -> new int[1],
(a, t) -> { a[0] += mapper.applyAsInt(t); },
(a, b) -> { a[0] += b[0]; return a; },
a -> a[0], CH_NOID);
}
summingInt比较复杂，不管只要看明白收集器的流程图，应该比较容易懂。把CollectorImpl中4个方法的职能连贯在一起。就是
- 准备应该数组当容器：
new int[1] - 将元素转换成Int并同元素中的值累加
- 将容器合并成1个值（放在某一个容器中）
- 返回这个值（最后的容器）
CH_NOID标识CollectorImpl的特性前面有讲过
- groupingBy()
这个方法是对一组序列进行分组的收集器（Collector），会根据classifier进行分类，让后交给downstream继续汇聚操作，最终返回（Map<K,D>）K 与D是泛型，具体如下
public static <T, K, D, A, M extends Map<K, D>> Collector<T, ?, M> groupingBy(Function<? super T, ? extends K> classifier,
Supplier<M> mapFactory,
Collector<? super T, A, D> downstream) {
Supplier<A> downstreamSupplier = downstream.supplier();
BiConsumer<A, ? super T> downstreamAccumulator = downstream.accumulator();
BiConsumer<Map<K, A>, T> accumulator = (m, t) -> {
K key = Objects.requireNonNull(classifier.apply(t), "element cannot be mapped to a null key");
A container = m.computeIfAbsent(key, k -> downstreamSupplier.get());
downstreamAccumulator.accept(container, t);
};
BinaryOperator<Map<K, A>> merger = Collectors.<K, A, Map<K, A>>mapMerger(downstream.combiner());
@SuppressWarnings("unchecked")
Supplier<Map<K, A>> mangledFactory = (Supplier<Map<K, A>>) mapFactory;
if (downstream.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)) {
return new CollectorImpl<>(mangledFactory, accumulator, merger, CH_ID);
}
else {
@SuppressWarnings("unchecked")
Function<A, A> downstreamFinisher = (Function<A, A>) downstream.finisher();
Function<Map<K, A>, M> finisher = intermediate -> {
intermediate.replaceAll((k, v) -> downstreamFinisher.apply(v));
@SuppressWarnings("unchecked")
M castResult = (M) intermediate;
return castResult;
};
return new CollectorImpl<>(mangledFactory, accumulator, merger, finisher, CH_NOID);
}
}
先从泛型开始分析
@param <T> 流中元素
@param <K> 分组的key（组名）类型
@param <A> 下游收集器的时候，收集器的中间类型
@param <D> 下有分组的返回结果
@param <M> 分组的结果类型
classifier不用说是分类器，mapFactory其实是结果容器，是一个Map，这里可以使用TreeMap和HashMap，可以自定义。
下游收集器是什么呢？
如图，下游收集器就是这个参数，用于对分组的结果再分组
如果对分组结果无顺序要求的话，可以使用groupByConcurrent，数据量大的时候，速度更快
操作步骤
1.获取下游收集器的容器方法
2.获取下游收集器的累加器downstreamAccumulator对象
3.借助下游收集器，构造新的累加器accumulator对象（分类器处理元素，生产K类型的key，使用computeIfAbsent产生中间类型container，使用下游收集器的累加器downstreamAccumulator处理key，和新的value）
4.创建一个合并器merger，实现两个Map合并
5.创建结果容器mangledFactory（强制转换，因为中间类型是Map<K,A>需要思考为什么可以转换
6,判断下游收集器有没有Finisher，有的话，就可以创建新的Collector并返回了
7.若有Finisher，就需要将Finisher换成下游收集器的Finisher再构建Collector。
return new CollectorImpl<>(mangledFactory, accumulator, merger, finisher, CH_NOID);
computeIfAbsent：若Map中存在key，则返回值，若不存在，则通过function构造一个（key-value），并返回value
构造accumulator的步骤略微复杂，因为结合了下游收集器，所以只需要把下游收集器downstreamAccumulator的处理，当做自己的accumulator的一个操作步骤就容易理解多了，combiner直接就是merge下游的combiner 调用了内部实现的mapMerger()方法。不是很难，可以自己看一下。
- 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());
}
这段代码不容易理解，mapping会对数据进行类型转换，源码中有一段示例：
Map<City, Set<String>> lastNamesByCity
= people.stream().collect(groupingBy(Person::getCity,
mapping(Person::getLastName, toSet())));
这段代码有一个groupingBy在前，mapping，在groupingBy内.
此处卡住1w年
根据前面groupBy知道，这里的mapping发挥的作用就是downloadStream下游收集器。如果你没有理解的话，参考Stream本身的Map就应该知道，这是吧把原始元素转换成新的元素后的收集器。
ps: 不能再详细的说了，说完比较唠叨。后面也算是给看客留个练习
pps: 注意一下finisher,也是很有意思的
本文写了近1年（我是有多懒），实际花费时间超过1周。如果你没有Java8基础，请翻看我N年前的文章
1.函数式接口和Java8的lambda表达式
2.函数式接口和Lambda表达式深入理解
3.Java8 Stream系列（一）从入坑到沉迷
4.Java8 Stream 系列（二）Stream应当注意的点
阅读量大的都是小白文，这篇应该看得人不多吧😝
喜欢请点个赞
转载请注明出处：https://www.jianshu.com/u/4915ed24d1e3
如有错误，请勿吝指正。谢谢
我的博客：https://xzing.github.io/
