java8 并行流原理及其使用

迈不过友情╰ 2024-04-18 23:26 66阅读 0赞

原理：  
    当线程在多核运行时，比如每一个核分配一个线程，当某一个线程阻塞，另外其他线程继续处理任务，当其他处理完了，此阻塞线程才刚开始执行任务任务，造成内存利用不足。java8在此做了相当大的改进，分为分拆个很多个小任务，在多核内存上执行，当其他线程任务队列执行完成，会偷偷的去拿阻塞队列中未执行完的进行执行，达到充分利用内存资源，高效的目的  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM4MDY4ODEy_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM4MDY4ODEy_size_16_color_FFFFFF_t_70 1]  
代码示例：

public class ForkJoinCaculate  extends RecursiveTask<Long> { 
    
        private  int start ;
        private  int end ;
    
        private final int THRESHOLD = 5000;
    
        public ForkJoinCaculate(int start, int end) { 
            this.start = start;
            this.end = end;
        }
    
        @Override
        protected Long compute() { 
            int length = end - start;
            long sum = 0;
            if(length < THRESHOLD){ 
                for (int i = 0; i < length; i++) { 
                    sum += i;
                }
                return  sum;
            }else{ 
                int middle = (start + end) / 2;
                ForkJoinCaculate left = new ForkJoinCaculate(start,middle);
                left.fork();
                // 拆分子任务，同时压入线程队列
                ForkJoinCaculate right = new ForkJoinCaculate(middle + 1,end);
                left.fork();
                return   left.join() + right.join();
            }
    
        }
    }

public class ForkJoinCaculateTest { 
        @Test
        public  void test1(){ 
            Instant start = Instant.now();
            ForkJoinPool pool = new ForkJoinPool();
            ForkJoinCaculate caculate = new ForkJoinCaculate(0,1000000);
            Long sum = pool.invoke(caculate);
            Instant end = Instant.now();
            System.out.println(String.format("返回值：%s,耗时：%s",sum, Duration.between(start,end).getSeconds()));
        }
        @Test
        public  void test2(){ 
            Instant start = Instant.now();
            long sum = LongStream.rangeClosed(0, 1000000000)
                    .parallel()
                    .reduce(0, Long::sum);
            Instant end = Instant.now();
            System.out.println(String.format("返回值：%s,耗时：%s",sum, Duration.between(start,end).getSeconds()));
        }
    
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM4MDY4ODEy_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/6092e4e202d0419fa0bd6d2aff4d56f8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM4MDY4ODEy_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/97ef1c3b0748438a889453cecc7f4bb7.png