1. 收集器
2. 预定义收集器
3. 广义的归约汇总
4. 并行流与ForkJoin框架

收集器

预定义收集器

1. Collectors.counting(),无参数,返回流元素的个数, 还可以直接对stream()对象使用.count(), 这个方法本身就是终端操作
2. Collectors.maxBy()和，Collectors.minBy(), 参数是是一个Comparator.
3. Collectors.summingInt(), 参数是一个把流元素映射为int的函数, 用于汇总. 对于Long和Double类型也有类型特化的Collectors.summingLong和Collectors.summingDouble
4. Collectors.averagingInt(),参数也是一个把流元素映射为int的函数, 求平均值. 同样有类型特化版本.
5. Collectors.summarizingInt(),参数也是一个把流元素映射为int的函数, 这个方法会一次性求出最大, 最小, 平均值, 求和, 个数, 结果是一个IntSummaryStatistics对象, 可以取出上边的结果. 同样也有类型特化版本.
6. Collectors.joining(), 无参数的版本, 是将流元素中每个对象的toString()的结果连接起来, 有一个重载的版本可以传入字符串参数当成分隔符.
7. Collectors.joining(), 无参数的版本, 是将流元素中每个对象的toString()的结果连接起来, 有一个重载的版本可以传入字符串参数当成分隔符.
8. Collectors.groupingBy, 下边详细讲述
9. Collectors.partitioningBy, 本质上就是按照布尔值的分组, 下边详细讲述
10. Collectors.toSet(), 收集成集合, 自动去重
11. Collectors.toCollection(), 收集成指定的集合, 使用在collect()方法内部的时候, 第二个参数要传一个集合类型的构造器, 比如menuStream.collect(toCollection(), ArrayList::new);

public static <T, K> Collector<T, ?, Map<K, List<T>>> groupingBy(Function<? super T, ? extends K> classifier) {
    return groupingBy(classifier, toList());
}

广义的归约汇总

并行流

public class TestTime {

    public static long measureSumPerf(Function<Long, Long> adder, long n) {
        long fastest = Long.MAX_VALUE;
        for (int i = 0; i < 10; i++) {
            long start = System.nanoTime();
            long sum = adder.apply(n);
            long duration = (System.nanoTime() - start) / 1_000_000;
            System.out.println("Result: " + sum);
            if (duration < fastest) fastest = duration;
        }
        return fastest;
    }

    public static long noSideEffectSum(long n) {
        Accumulator accumulator = new Accumulator();
        LongStream.rangeClosed(1, n).forEach(accumulator::add);
        return accumulator.total;
    }

    public static long sideEffectParallelSum(long n) {
        Accumulator accumulator = new Accumulator();
        LongStream.rangeClosed(1, n).parallel().forEach(accumulator::add);
        return accumulator.total;
    }

    public static void main(String[] args) {
        //多线程竞争的情况
        System.out.println("SideEffect parallel sum done in: " + measureSumPerf(TestTime::sideEffectParallelSum, 10_000_000L) + "msecs");

        //单线程处理,不会有竞争的情况
        System.out.println("SideEffect parallel sum done in: " + measureSumPerf(TestTime::noSideEffectSum, 10_000_000L) + "msecs");
    }

}

import java.util.concurrent.RecursiveTask;

//编写任务, 继承java.util.concurrent.RecursiveTask<Long>, 这是ForkJoin框架的任务, 是一个Future<V>接口的实现类
//这个接口只有一个抽象方法, 叫做compute(), 返回泛型类型

public class ForkJoinSumCalculator extends RecursiveTask<Long> {

    //要操作一个数组, 由于不能多线程竞争, 因此引入数组的开始与结束的变量, 以合理拆分数组给各个线程
    private final long[] numbers;
    private final int start;
    private final int end;

    //不拆分的最小值
    public static final long THRESHOLD = 10_000;

    //私有构造器, 由于仅仅是在内部创建线程, 因此不暴露给外界
    private ForkJoinSumCalculator(long[] numbers, int start, int end) {
        this.numbers = numbers;
        this.start = start;
        this.end = end;
    }

    //工具方法, 用于汇总numbers数组指定的区域
    private long computeSequentially() {
        long sum = 0;
        for (int i = start; i < end; i++) {
            sum += numbers[i];

        }
        return sum;
    }

    //核心的compute方法, 将当前的任务分解为两个任务, 分解方法是分解索引, 然后将分解后的索引和同一个数组对象分成两个任务
    @Override
    protected Long compute() {
        int length = end - start;
        if (length <= THRESHOLD) {
            return computeSequentially();
        }
        //左半边任务
        ForkJoinSumCalculator leftTask =
                new ForkJoinSumCalculator(numbers, start, start + length / 2);
        //左半边任务交给新的线程
        leftTask.fork();
        //右半边任务, 直接在当前线程内启动
        ForkJoinSumCalculator rightTask =
                new ForkJoinSumCalculator(numbers, start + length / 2, end);
        Long rightResult = rightTask.compute();
        //等待左半边任务的结束, 然后合并结果
        Long leftResult = leftTask.join();
        return leftResult + rightResult;
    }

    //静态方法用于启动任务
    public static long forkJoinSum(long n) {
        long[] numbers = LongStream.rangeClosed(1, n).toArray();
        //创建一个最初的任务, 然后交给ForkJoinPool()来执行
        ForkJoinTask<Long> task = new ForkJoinSumCalculator(numbers, 0, numbers.length);
        return new ForkJoinPool().invoke(task);
    }


    public static void main(String[] args) {
        long result = forkJoinSum(999);
        System.out.println(result);
    }
}