欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

JDK1.8新特性 ForkJoin并行流的使用(三)

程序员文章站 2022-02-06 11:31:54
...

在jdk1.8新的stream包中针对集合的操作也提供了并行操作流和串行操作流。并行流就是把内容切割成多个数据块,并且使用多个线程分别处理每个数据块的内容。Stream api中声明可以通过parallel()与sequential()方法在并行流和串行流之间进行切换。
jdk1.8并行流使用的是fork/join框架进行并行操作。
Fork/Join 框架:就是在必要的情况下,将一个大任务,进行拆分(fork)成若干个小任务(拆到不可再拆时),再将一个个的小任务运算的结果进行 join 汇总。
采用 “工作窃取”模式(work-stealing):
当执行新的任务时它可以将其拆分分成更小的任务执行,并将小任务加到线 程队列中,然后再从一个随机线程的队列中偷一个并把它放在自己的队列中 相对于一般的线程池实现,fork/join框架的优势体现在对其中包含的任务的 处理方式上.在一般的线程池中如果一个线程正在执行的任务由于某些原因 无法继续运行,那么该线程会处于等待状态.而在fork/join框架实现中,如果 某个子问题由于等待另外一个子问题的完成而无法继续运行.那么处理该子 问题的线程会主动寻找其他尚未运行的子问题来执行.这种方式减少了线程

public class ForkJoin extends RecursiveTask<Long> {

    private static final long serialVersionUID = 23423422L;

    private long start;
    private long end;

    // 定义阙值
    private static final long THRESHOLD = 10000L;

    public ForkJoin() {
    }

    public ForkJoin(long start, long end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        if (end - start <= THRESHOLD) {
            long sum = 0;
            for (long i = start; i < end; i++) {
                sum += i;
            }
            return sum;
        } else {
            long middle = (end - start) / 2;
            ForkJoin left = new ForkJoin(start, middle);
            //拆分子任务,压入线程队列
            left.fork();
            ForkJoin right = new ForkJoin(middle + 1, end);
            right.fork();

            //合并并返回
            return left.join() + right.join();
        }
    }

    public static void main(String args[]){
        /**
         * 使用ForkJoin方式
         */
        //开始时间
/*        Instant startTime=Instant.now();
        ForkJoinPool forkJoinPool=new ForkJoinPool();

        ForkJoinTask<Long> forkJoinTask=new ForkJoin(1L, 10000000000L);
        //没有返回值的 pool.execute();
        //有返回值
        long sum=forkJoinPool.invoke(forkJoinTask);
        System.out.println("返回值为:"+sum);
        Instant endTime=Instant.now();
        System.out.println(Duration.between(startTime,endTime).getSeconds());*/


        /**
         * 使用parallel
         */
        //开始时间
        Instant startTime=Instant.now();
        LongStream.rangeClosed(0L,100000000000L).parallel().reduce(0,Long::sum);
        //结束时间
        Instant endTime=Instant.now();
        System.out.println(Duration.between(startTime,endTime).getSeconds());

    }
}
相关标签: Java JDK1.8