Jdk8--新特性--串并行流与ForkJoin框架

Love The Way You Lie 2022-05-19 00:50 206阅读 0赞

并行流就是把一个内容分成多个数据块，并用不同的线程分别处理每个数据块的流。穿行流则相反，并行流的底层其实就是ForkJoin框架的一个实现。

那么先了解一下ForkJoin框架吧。

Fork/Join框架：在必要的情况下，将一个大任务，进行拆分（fork） 成若干个子任务（拆到不能再拆，这里就是指我们制定的拆分的临界值），再将一个个小任务的结果进行join汇总。

Fork/Join与传统线程池的区别！

Fork/Join采用“工作窃取模式”，当执行新的任务时他可以将其拆分成更小的任务执行，并将小任务加到线程队列中，然后再从一个随即线程中偷一个并把它加入自己的队列中。

就比如两个CPU上有不同的任务，这时候A已经执行完，B还有任务等待执行，这时候A就会将B队尾的任务偷过来，加入自己的队列中，对于传统的线程，ForkJoin更有效的利用的CPU资源！

我们来看一下ForkJoin的实现：实现这个框架需要继承RecursiveTask 或者 RecursiveAction ，RecursiveTask是有返回值的，相反Action则没有

测试

import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.ForkJoinTask;
    import java.util.stream.LongStream;
    
    import org.junit.Test;
    
    public class TestForkJoin {
    	
    	@Test
    	public void test1(){
    		long start = System.currentTimeMillis();
    		
    		ForkJoinPool pool = new ForkJoinPool();
    		ForkJoinTask<Long> task = new ForkJoinCalculate(0L, 10000000000L);
    		
    		long sum = pool.invoke(task);
    		System.out.println(sum);
    		
    		long end = System.currentTimeMillis();
    		
    		System.out.println("耗费的时间为: " + (end - start)); //112-1953-1988-2654-2647-20663-113808
    	}
    	
    	@Test
    	public void test2(){
    		long start = System.currentTimeMillis();
    		
    		long sum = 0L;
    		
    		for (long i = 0L; i <= 10000000000L; i++) {
    			sum += i;
    		}
    		
    		System.out.println(sum);
    		
    		long end = System.currentTimeMillis();
    		
    		System.out.println("耗费的时间为: " + (end - start)); //34-3174-3132-4227-4223-31583
    	}
    	
    	@Test
    	public void test3(){
    		long start = System.currentTimeMillis();
    		
    		Long sum = LongStream.rangeClosed(0L, 10000000000L)
    							 .parallel()
    							 .sum();
    		
    		System.out.println(sum);
    		
    		long end = System.currentTimeMillis();
    		
    		System.out.println("耗费的时间为: " + (end - start)); //2061-2053-2086-18926
    	}
    
    }

import java.util.concurrent.RecursiveTask;
    
    public class ForkJoinCalculate extends RecursiveTask<Long>{
    
    	/**
    	 * 
    	 */
    	private static final long serialVersionUID = 13475679780L;
    	
    	private long start;
    	private long end;
    	
    	private static final long THRESHOLD = 10000L; //临界值
    	
    	public ForkJoinCalculate(long start, long end) {
    		this.start = start;
    		this.end = end;
    	}
    	
    	@Override
    	protected Long compute() {
    		long length = end - start;
    		
    		if(length <= THRESHOLD){
    			long sum = 0;
    			
    			for (long i = start; i <= end; i++) {
    				sum += i;
    			}
    			
    			return sum;
    		}else{
    			long middle = (start + end) / 2;
    			
    			ForkJoinCalculate left = new ForkJoinCalculate(start, middle);
    			left.fork(); //拆分，并将该子任务压入线程队列
    			
    			ForkJoinCalculate right = new ForkJoinCalculate(middle+1, end);
    			right.fork();
    			
    			return left.join() + right.join();
    		}
    		
    	}
    
    }

我们观察上面可以看出来执行10000000000L的相加操作各自执行完毕的时间不同。观察到当数据很大的时候ForkJoin比普通线程实现有效的多，但是相比之下ForkJoin的实现实在是有点麻烦，这时候Java 8 就为我们提供了一个并行流来实现ForkJoin实现的功能。可以看到并行流比自己实现ForkJoin还要快

Java 8 中将并行流进行了优化，我们可以很容易的对数据进行并行流的操作，Stream API可以声明性的通过parallel()与sequential()在并行流与穿行流中随意切换！