Java8 Stream 系列（三）深入源码之Collector

Stream 中有两个操作比较繁琐，一个是reduce，另一个是collect
其中，collect的理解相对难一些，所以先研究collect，（汇聚操作）

0x00

1x00：Collector接口

1	<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）保证了串行和并行执行结果的等价性

同一性
1
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::new

BiConsumer<A, T> accumulator(); 将元素合并到结果容器中 List::add

BinaryOperator<A> combiner(); 将两个结果容器合并(left, right) -> { left.addAll(right); return left; }
。

toCollection(Supplier 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展开之后，即可统计数量。很方便
1
2
3
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。

1	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会对数据进行类型转换，源码中有一段示例：

1
2
3

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/