7、java8并行流和串行流

标签：行操作   loading   汇总   否则   inf   就是   不可   private   key   

1、概念

并行流就是把一个内容分成多个数据块，并用不同的线程分别处理每个数据块的流

Java8中将并行进行了优化，我们可以很容易的对数据进行并行操作。

Stream API可以申明性的通过parallel()与sequential()在并行流与顺序流之间进行切换

2、Fork/Join框架

2.1、概念

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

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

采用“工作窃取”模式（work-stealing）：

当执行新的任务时，它可以将其拆分为更小的任务执行，并将小任务加到线程队列中，然后再从一个随机线程的队列中偷一个并把它放在自己的队列中。

相对于一般的线程池实现：

• fork/join框架的优势体现在对其中包含的任务的处理方式上，在一般的线程池中，如果一个线程正在执行的任务由于某些原因无法继续运行，那么该线程会处于等待状态

• 在fork/join框架实现中，如果某个子问题由于等待另外一个子问题的完成而无法继续运行，那么处理该子问题的线程就会主动寻找其它尚未运行的子问题来执行，这种方式减少了线程等待时间，提高了性能。

2.3、代码示例

package java8;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;

/**
 * @Description: forkjoin框架demo
 * @Date: 2020/8/15 23:57
 * @Auther: zhaodi
 */
@Slf4j
public class ForkJoin extends RecursiveTask<Long> {
    // 任务划分的细粒度
    private static Long MAX = 200l;
    // 子任务开始计算的值
    private Long startVal;
    // 子任务结束计算的值
    private Long endVal;
    public ForkJoin(Long startVal, Long endVal) {
        this.startVal = startVal;
        this.endVal = endVal;
    }
    @Override
    protected Long compute() {
        // 如果条件成立，说明这个任务所需要计算的数值分为足够小了
        // 可以正式进行累加计算了
        if (endVal - startVal < MAX) {
            log.info("开始计算，startVal:{},endVal:{}", startVal, endVal);

            Long sum = 0l;
            for (long i = this.startVal; i <= this.endVal; i++) {
                sum += i;
            }
            return sum;
            // 否则再进行任务拆分，拆分成两个任务
        } else {
            ForkJoin taskLeft = new ForkJoin(startVal, (startVal + endVal) / 2);
            taskLeft.fork();
            ForkJoin taskRight = new ForkJoin((startVal + endVal) / 2, endVal);
            taskRight.fork();
            return taskLeft.join() + taskRight.join();
        }
    }
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ForkJoinPool pool = new ForkJoinPool();
        ForkJoinTask<Long> t = pool.submit(new ForkJoin(1l,1000000000L));
        Long result = t.get();
        System.out.println(result); //504194303981605715
    }
}

3、Java8并行流

Fork/Join在实际编码中还是很少使用的，因为使用起来还是比较复杂的,通过上面的代码我们就可以发现，用起来的难度系数还是比较大的。

但是我们就没有更好的办法解决这个问题吗？答案是肯定有的，在Java8中我们提供了parallel（）底层就是用Fork/Join框架实现的

/**
 * @Description:
 * @Date: 2020/8/12 23:16
 * @Auther: zhaodi
 */
public class ParalleStreamDemo {

    public static void main(String[] args) {
        List<Student> studentList = new ArrayList<>();
        studentList.add(new Student(1, "赵迪", 23));
        studentList.add(new Student(3, "tom", 25));
        studentList.add(new Student(2, "tom", 27));
        studentList.add(new Student(2, "rose", 22));
        studentList.parallelStream().filter(x->x.getAge()>23).collect(Collectors.toList());
    }
}

7、java8并行流和串行流

标签：行操作   loading   汇总   否则   inf   就是   不可   private   key   

原文地址：https://www.cnblogs.com/zhaod/p/13514669.html