Java8新特性之并行流与串行流

并行流与串行流

      并行流就是把一个内容分红多个数据块，并用不一样的线程分别处理每一个数据块的流。
      Java 8 中将并行进行了优化，咱们能够很容易的对数据进行并行操做。Stream API 能够声明性地经过parallel() 与sequential() 在并行流与顺序流之间进行切换。

Fork/Join 框架

       Fork/Join 框架：就是在必要的状况下，将一个大任务，进行拆分(fork)成若干个小任务（拆到不可再拆时），再将一个个的小任务运算的结果进行join 汇总.

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

      采用“工做窃取”模式（work-stealing）：当执行新的任务时它能够将其拆分分红更小的任务执行，并将小任务加到线程队列中，而后再从一个随机线程的队列中偷一个并把它放在本身的队列中。

     相对于通常的线程池实现,fork/join框架的优点体如今对其中包含的任务的处理方式上.在通常的线程池中,若是一个线程正在执行的任务因为某些缘由没法继续运行,那么该线程会处于等待状态.而在fork/join框架实现中,若是某个子问题因为等待另一个子问题的完成而没法继续运行.那么处理该子问题的线程会主动寻找其余还没有运行的子问题来执行.这种方式减小了线程的等待时间,提升了性能.

ForkJoinCalculate.java

package www.muzi.com;

import java.util.concurrent.RecursiveTask;

/**
 * Date：2017/3/10 15:18
 */
public class ForkJoinCalculate extends RecursiveTask<Long>{

	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();
		}
	}
}

TestForkJoin.java

package www.muzi.com;

import org.junit.Test;

import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.stream.LongStream;

/**
 * Date：2017/3/10 17:12
 */
public class TestForkJoin {
	Long num = 10000000000L;
	/**
	 * ForkJoin
	 */
	@Test
	public void test1(){
		Instant start = Instant.now();
		ForkJoinPool pool = new ForkJoinPool();
		ForkJoinTask<Long> task = new ForkJoinCalculate(0,num);
		Long sum = pool.invoke(task);
		System.out.println(sum);
		Instant end = Instant.now();
		System.out.println("耗费时间：" + Duration.between(start,end).toMillis());//3904
	}

	/**
	 * 普通for循环
	 */
	@Test
	public void test2(){
		Instant start = Instant.now();
		Long sum = 0L;
		for (int i = 0; i < num; i++) {
			sum += i;
		}
		System.out.println(sum);
		Instant end = Instant.now();
		System.out.println("耗费时间：" + Duration.between(start,end).toMillis());//由于时间太长。。。。不等了
	}

	/**
	 * 并行流
	 */
	@Test
	public void test3(){
		Instant start = Instant.now();
		LongStream.rangeClosed(0, num)
				.parallel()
				.reduce(0, Long:: sum);
		Instant end = Instant.now();
		System.out.println("耗费时间：" + Duration.between(start,end).toMillis());//3887
	}

	/**
	 * 串行流
	 */
	@Test
	public void test4(){
		Instant start = Instant.now();
		LongStream.rangeClosed(0, num)
				.sequential()
				.reduce(0, Long:: sum);
		Instant end = Instant.now();
		System.out.println("耗费时间：" + Duration.between(start,end).toMillis());//7398
	}
}