收集器源码解析

一 概述

 Collectors提供了常见的工具收集器，其中基本分为两种：一种是借助reducing方法，另一种是构造CollectorImpl实例。

二 reducing方案

  方法签名：
Collector<T, ?, U> reducing(U identity,Function<? super T, ? extends U> mapper, BinaryOperator< U > op)
  泛型含义：
   T 元素的类型
   U 收集器的结果类型
  方法含义：
   根据mapper和op对流中的元素执行汇聚操作
  参数含义：
   identity 汇聚操作的初始值
   mapper 对流中的元素执行转换操作，将T类型转为U类型
   op 聚合mapper转换后的值
  源码分析：

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);}
    
构造的是CollectorImpl，生成的Collector的元素类型是T，结果类型是U

使用的CollectorImpl的构造方法是 
    CollectorImpl(Supplier<A> supplier,
                      BiConsumer<A, T> accumulator,
                      BinaryOperator<A> combiner,
                      Function<A,R> finisher,
                      Set<Characteristics> characteristics)
    
    supplier是构造中间收集器
    accumulator是收集器的中间累积器--->聚合元素
    combiner是中间收集器的合并器
    finisher是将中间收集器转为最终的结果
    characteristics是收集器的特征
    因此核心就是配置以上的参数，理清泛型
    
boxSupplier(identity)的作用是构造supplier。
    boxSupplier(identity)的源码如下
     private static <T> Supplier<T[]> boxSupplier(T identity) {
        return () -> (T[]) new Object[] { identity };//第二处
    }
    上述的T就是传入的参数类型，也就是第一处的identity的U类型。第二处发现是
    生成了一个元素的数组，数组中的元素就是identity。
综上，构造的CollectorImpl的supplier就是返回U类型数组：()->U[0]

第三处： (a, t) -> { a[0] = op.apply(a[0], mapper.apply(t));
    第三处实际构造的是BiConsumer，中间结果的收集器。a就是中间收集器的类型，也就是第一处构造的类型U[0]，
t就是待累计元素的类型T，这个lambda体就是对数组中的identity不断的赋指，执行流程就是首先对元素t执行mapper
操作，调用op进行累积（不断的对identity赋值）。
第四处： (a, b) -> { a[0] = op.apply(a[0], b[0]); return a;}
    此处构造的是并行模式的合并器，将分段结果合并。
第五处：a -> a[0] 将中间结果转为最终结果，提取出元素的类型（identity的类型U）
第六处：CH_NOID：emptySet
    因为需要执行finisher的lambda体，因此特征中肯定不能包含IDENTITY_FINISH。
    

以counting为例分析reduceing
public static <T> Collector<T, ?, Long> counting() {
        return reducing(0L, e -> 1L, Long::sum);
    }
此处参数的含义：0L为identity，e -> 1L为mapper，Long::sum为op
根据生成的CollectorImpl(
                boxSupplier(0L),
                (a, t) -> { a[0] = a[0]+ (e -> 1L); },
                (a, b) -> { a[0] = a[0]+b[0]; return a; },
                a -> a[0], CH_NOID);
    boxSupplier(0L)->  会生成一个Long类型的数组，数组的元素就是0
    accumulator-> 会将元素转为1，然后将数组的元素加1
    combiner-> 将每次分段累积的结果相加
    finisher-> 就是取出数组的元素 完成了求和

分组

 分组的函数式groupingBy，JDK提供了三个重载

//第一处
groupingBy(Function<? super T, ? extends K> classifier)
    对流中的T类型元素实现了分组聚合，分组的依据是classifier，分组的结果就是Map。
classifier将T类型的元素，转为Map中key的K类型。这种收集器中，Map的key所对应的value是List，
也就是结果类型是Map<K,List<T>>

//第二处
groupingBy(Function<? super T, ? extends K> classifier,Collector<? super T, A, D> downstream)
    可以实现流中元素的级联分组，classifier对流中元素确定Key，每个key所对应的value使用downstream再次聚合
classifier将T类型的元素转为K类型，downstream操作T类型的元素，其结果类型是D，因此此分组的结果累型就是Map<K,D>

//第三处
groupingBy(Function<? super T, ? extends K> classifier,Supplier<M> mapFactory,Collector<? super T, A, D> downstream)
    mapFactory的作用就是提供定制的Map

 分组的收集器的结果类型就是Map，Map需要确定的是key，value以及key与value之间的关系。
 

  第一处

 public static <T, K> Collector<T, ?, Map<K, List<T>>>
    groupingBy(Function<? super T, ? extends K> classifier) {
        return groupingBy(classifier, toList());
    }
    
    泛型含义：
        T 元素的类型
        K Map中key的类型
    Map的key就是classifier函数生成的
    value就是元素中根据key筛选出的list
    key和value的关系就是 K----List<T>
    

  第二处

public static <T, K, A, D>
    Collector<T, ?, Map<K, D>> groupingBy(Function<? super T, ? extends K> classifier,Collector<? super T, A, D> downstream) {
        return groupingBy(classifier, HashMap::new, downstream);
    }

    泛型含义：
        T 流中元素的类型
        K Map中key的类型
        A downstream的中间收集器的类型
        D downstream的结果类型
        
    Map的key就是classifier函数生成的 
    生成的Map就是HashMap
    value就是downstream的汇聚结果

  第三处

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
        Supplier<A> downstreamSupplier = downstream.supplier();
        //下流的accumulator，类型是A是中间类型，T是元素类型
        //value的汇聚行为
        BiConsumer<A, ? super T> downstreamAccumulator = downstream.accumulator();
        //生成的Map，Map的积累器
        //解决如何添加元素
        BiConsumer<Map<K, A>, T> accumulator = (m, t) -> {
            //key不为空
            K key = Objects.requireNonNull(classifier.apply(t), "element cannot be mapped to a null key");
            // computeIfAbsent 根据key得到value，
            // 如果value为空，则根据参数mappingFunction生成value，并将key和value添加至Map，并返回生成的value
            // 如果value不为空，则直接返回value
            // 此处确定了key和value，value就是从下游的supplier获取
            A container = m.computeIfAbsent(key, k -> downstreamSupplier.get());
            // 下游的聚合行为，将元素添加至value（value是容器）中
            downstreamAccumulator.accept(container, t);
        };
        BinaryOperator<Map<K, A>> merger = Collectors.<K, A, Map<K, A>>mapMerger(downstream.combiner());
        // 构造Map
        @SuppressWarnings("unchecked")
        Supplier<Map<K, A>> mangledFactory = (Supplier<Map<K, A>>) mapFactory;

        //如果下游特征包含了IDENTITY_FINISH（不会走finisherlambda体，认为中间收集器就是结果收集器）
        if (downstream.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)) {
            //构造CollectorImpl
            //supplier提供中间收集器（结果收集器）Map(K,A)
            //accumulator累积元素
            //merger合并子Map
            //特征是IDENTITY_FINISH
            return new CollectorImpl<>(mangledFactory, accumulator, merger, CH_ID);
        }else {//不包含IDENTITY_FINISH
            @SuppressWarnings("unchecked")
            Function<A, A> downstreamFinisher = (Function<A, A>) downstream.finisher();
            //实现finisher
            Function<Map<K, A>, M> finisher = intermediate -> {
                //对value替换
                intermediate.replaceAll((k, v) -> downstreamFinisher.apply(v));
                @SuppressWarnings("unchecked")
                M castResult = (M) intermediate;
                //将结果返回
                return castResult;
            };
            return new CollectorImpl<>(mangledFactory, accumulator, merger, finisher, CH_NOID);
        }
    }