> 文档中心 > 【函数式编程实战】(六) Stream高并发实战

【函数式编程实战】(六) Stream高并发实战

前言
📫 作者简介:小明java问道之路,专注于研究计算机底层,就职于金融公司后端高级工程师,擅长交易领域的高安全/可用/并发/性能的设计和架构📫 
🏆 Java领域优质创作者、阿里云专家博主、华为云享专家🏆
🔥 如果此文还不错的话,还请👍关注点赞收藏三连支持👍一下博主哦

导读

我们上讲看Stream接口提供大量API可以方便的处理元素,这讲Stream高并发(并发、并行、多线程)、ForkJoin线程池框架的实战

一、并行流(parallelStream、parallel、sequential)

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

parallelStream、.parallel()都可以将流转换成并行流,.parallel()的粒度更小。要注意的是顺序流( .stream) 调用parallel方法不意味本身有什么实际变化,它内部设置了一个boolean,表示调用parallel之后的操作都是并行的

对并行流调用后可以使用 .sequential() 变成顺序流。

 List collect10 = orderInfos.stream().parallel()  .map(OrderInfo::getOrderId).collect(toList()); BigDecimal totalSubOrderAmt1 = orderInfos.stream()  .map(OrderInfo::getSubOrderInfoList)  .flatMap(subOrderInfos -> subOrderInfos.stream())  .filter(subOrderInfo -> null != subOrderInfo.getSubOrderAmt())  .parallel()  .map(SubOrderInfo::getSubOrderAmt)  .sequential()  .reduce(BigDecimal.ZERO, BigDecimal::add); List collect11 = orderInfos.parallelStream()  .map(OrderInfo::getOrderId).collect(toList());

并行流 不是一定并行,多线程中保证原子操作会有对象的可变状态,当多个线程共享对象时,共享的可变状态会不断被线程锁住,会影响并行流和并行计算

所以我们是否可以使用并行流需要避免共享变量,而且当较小的数据量的时候并行处理的开销不一定会小于计算开销,同时集合的数据结构也会对并行流有影响,ArrayList拆分的开销,要小于LinkedList(前者底层是数组可以直接在内存拆分,后者底层是链表,内存不连续需要遍历拆分,ArrayList、HashSet、TreeSet可拆分性好,LinkedList可拆分性极差)

二、Fork/Jion框架

并行流的背后的原理是java7里面的分支/合并框架,分支/合并框架的目的就是以递归的方式,将数据块(任务)并行的拆分成更小的数据块,然后将每个子任务合并成整体,ExecutorService 接口的实现,ExecutorService把子任务分配给线程池 ForkJionPool 中的工作线程。

要把任务提交到这个ForkJionPool 线程池,必须创建RecursiveTask的实现类,要定义RecursiveTask只需要实现他的抽象方法compute

创建一个 ForkJoinTask,并把对象传给我们自定义的ForkJoinCalculator,创建一个 ForkJoinPool 并把任务传递给他的调用方法invoke,返回值就是ForkJoinCalculator定义的结果

扩展RecursiveTask,穿件ForkJoin框架,创建起始位置和终止位置,实现compute方法,实现fork、join金额累加

    public static void main(String[] args) { List orderInfos = Arrays.asList(new OrderInfo("123", BigDecimal.ONE),  new OrderInfo("456", BigDecimal.TEN), new OrderInfo("789", BigDecimal.TEN)); // 创建一个 ForkJoinTask,并把对象传给我们自定义的ForkJoinCalculator ForkJoinCalculator orderInfoForkJoinTask = new ForkJoinCalculator(orderInfos, 0, orderInfos.size()); // 创建一个 ForkJoinPool 并把任务传递给他的调用方法,返回值就是ForkJoinCalculator定义的结果 System.out.println(new ForkJoinPool().invoke(orderInfoForkJoinTask));    }    /**     * 扩展RecursiveTask,穿件ForkJoin框架,创建起始位置和终止位置     */    static class ForkJoinCalculator extends RecursiveTask { List orderInfos; int start; int end; BigDecimal amt; public ForkJoinCalculator(List orderInfos, int start, int end) {     this.orderInfos = orderInfos;     this.amt = orderInfos.get(start).getOrderAmt();     this.start = start;     this.end = end; } @Override protected BigDecimal compute() {     int length = end - start;     // 如果大小等于阈值,返回结果     if (length < 2)  return amt; // 订单金额     ForkJoinCalculator task = new ForkJoinCalculator(orderInfos, start, start + length / 2);     task.fork();     ForkJoinCalculator task1 = new ForkJoinCalculator(orderInfos, start + length / 2, end);     BigDecimal compute = task1.compute();     BigDecimal join = task.join();     System.out.println("compute:" + compute + "  join:" + join);     return compute.add(join); // 累加 }    }

Fork/Jion框架还有几个框架需要注意的,对于一个任务调用join方法,会阻塞调用方,直到出任务结果;不应该在RecursiveTask里面实现ForkJoinPool 应该使用compute或fork方法;compute实现需要中断方法并且这里面实现比较困难需要多加练习。

工作窃取,在实际工作中任务差不多被平均分配到ForkJoinPool的所有线程,每个线程都为分配给线程的任务保存一个 双向LinkedQueue(双向链式队列),每个人物完成就会从队列的头取一个在进行执行,但是某些线程执行可能过快,此时这个线程会从其他线程的 队列尾,取走一个任务执行,这也是Fork/Jion框架能有高性能的原因

总结

Stream高并发(并发、并行、多线程)、ForkJoin线程池框架的实战

88读书网