Java8 Stream 自定义Collector

在之前的例子中，我们都是使用Collectors的静态方法提供的CollectorImpl，为接口Collector<T, A, R>的一个实现类，为了自定义我们自己的Collector，先来分析一下Collector接口。
一、分析接口Collector

/**
 * @param <T> 要收集的元素的泛型
 * @param <A> 累加器容器的类型，
 * @param <R> 收集操作得到的对象类型
 * @author futao
 * @date 2020/9/24
 */
public class MyCustomCollector<T, A, R> implements Collector<T, A, R> {
    /**
     * 创建一个接收结果的可变容器
     *
     * @return a function which returns a new, mutable result container
     */
    @Override
    public Supplier<A> supplier() {
        return null;
    }

    /**
     * 将流中的元素放入可变容器中的逻辑, 方法
     *
     * @return a function which folds a value into a mutable result container
     */
    @Override
    public BiConsumer<A, T> accumulator() {
        return null;
    }

    /**
     * 组合结果，当流被拆分成多个部分时，需要将多个结果合并。
     *
     * @return a function which combines two partial results into a combined
     * result
     */
    @Override
    public BinaryOperator<A> combiner() {
        return null;
    }

    /**
     * 最后调用：在遍历完流后将结果容器A转换为最终结果R
     *
     * @return a function which transforms the intermediate result to the final
     * result
     */
    @Override
    public Function<A, R> finisher() {
        return null;
    }

    /**
     * 返回一个描述收集器特征的不可变集合，用于告诉收集器时可以进行哪些优化，如并行化
     *
     * @return an immutable set of collector characteristics
     */
    @Override
    public Set<Characteristics> characteristics() {
        return null;
    }
}

• 描述收集器的特征: enum Characteristics

enum Characteristics {
    /**
     * 意味着结果容器支持多线程并发操作accumulator()方法向容器中放入元素。
     * 如果收集器没有被同时标记为无序的`UNORDERED`，则该特征只在数据源为无序（如set）时才有效
     */
    CONCURRENT,

    /**
     * 规约操作不保证集合中元素的顺序
     * 如结果容器为Set的场景下
     */
    UNORDERED,

    /**
     * 表明`完成方法finisher`是一个恒等式，可以被忽略
     * 累加器的结果combiner A 将会作为最终的结果R
     * 这也要求，直接将A转换成R是安全的
     */
    IDENTITY_FINISH
}

• Collectors中常见的Characteristics组合

    static final Set<Collector.Characteristics> CH_CONCURRENT_ID
            = Collections.unmodifiableSet(EnumSet.of(Collector.Characteristics.CONCURRENT,
                                                     Collector.Characteristics.UNORDERED,
                                                     Collector.Characteristics.IDENTITY_FINISH));
    static final Set<Collector.Characteristics> CH_CONCURRENT_NOID
            = Collections.unmodifiableSet(EnumSet.of(Collector.Characteristics.CONCURRENT,
                                                     Collector.Characteristics.UNORDERED));
    static final Set<Collector.Characteristics> CH_ID
            = Collections.unmodifiableSet(EnumSet.of(Collector.Characteristics.IDENTITY_FINISH));
    static final Set<Collector.Characteristics> CH_UNORDERED_ID
            = Collections.unmodifiableSet(EnumSet.of(Collector.Characteristics.UNORDERED,
                                                     Collector.Characteristics.IDENTITY_FINISH));
    static final Set<Collector.Characteristics> CH_NOID = Collections.emptySet();

• Collector执行规约过程

图片来源:《java8 in action》

二、自定义一个功能与Collectors.toList()一致的Collector

/**
 * 自定义收集器
 *
 * @author futao
 * @date 2020/9/24
 */
public class MyCollectors {

    private MyCollectors() {
    }

    /**
     * 描述：将流中的元素转换成List<T>输出
     *
     * Collector三个泛型:
     *                  <入参泛型类型，
     *                  中间结果容器类型(在这个例子中为List<T>)，
     *                  最终结果容器类型(这个例子中也是List<T>)>
     * @param <T>
     */
    public static class ToList<T> implements Collector<T, List<T>, List<T>> {

        /**
         * 创建一个接收结果的可变容器
         * 即：创建一个List<T>对象的方法
         *
         * @return
         */
        @Override
        public Supplier<List<T>> supplier() {
            return ArrayList::new;
        }

        /**
         * 将流中的元素放入可变容器中的方法
         *
         * @return
         */
        @Override
        public BiConsumer<List<T>, T> accumulator() {
            // return (list, item) -> list.add(item);
            return List::add;
        }

        /**
         * 组合结果，当流被拆分成多个部分时，需要将多个结果合并。
         *
         * @return
         */
        @Override
        public BinaryOperator<List<T>> combiner() {
            return (list1, list2) -> {
                list1.addAll(list2);
                return list1;
            };
        }


        /**
         * 最后调用：在遍历完流后将结果容器A转换为最终结果R
         * 在该例子中，combiner结果可作为最终结果，所以返回一个恒等式
         *
         * @return
         */
        @Override
        public Function<List<T>, List<T>> finisher() {
            // return x -> x;
            return Function.identity();
        }

        /**
         * 返回一个描述收集器特征的不可变集合
         * 该例子中可用的特性是：
         *                  finisher可跳过，直接将combiner结果返回。
         *                  需要保证有序
         *                  不可并发
         *
         * @return
         */
        @Override
        public Set<Characteristics> characteristics() {
            return Collections.unmodifiableSet(EnumSet.of(Characteristics.IDENTITY_FINISH));
        }
    }
}

• 测试

public static void main(String[] args) {
    Apple chinaApple = new Apple(10, "中国");
    Apple usApple = new Apple(20, "米国");
    Apple koreaApple = new Apple(30, "韩国");
    Apple japanApple = new Apple(40, "日本");

    List<Apple> collect = Stream.of(chinaApple, usApple, koreaApple, japanApple)
            // 使用自定义的收集器
            .collect(new MyCollectors.ToList<>());
    System.out.println(JSON.toJSONString(collect, true));
}

• 结果

[
    {
        "country":"中国",
        "weight":10
    },
    {
        "country":"米国",
        "weight":20
    },
    {
        "country":"韩国",
        "weight":30
    },
    {
        "country":"日本",
        "weight":40
    }
]

三、 当finisher为恒等式时的简便写法

    List<Apple> simpleListApples = Stream.of(chinaApple, usApple, koreaApple, japanApple)
            // 对于finisher为恒等式，可以用这种简单的写法
            .<List<Apple>>collect(ArrayList::new, List::add, List::addAll);

    System.out.println(JSON.toJSONString(simpleListApples, true));

• 使用这种方式定义的Collector收集器默认的特性是IDENTITY_FINISH和CONCURRENT组合，当数据源为无序集合时，CONCURRENT才会体现出作用。