外墙铝单板厂家欢迎您的咨询来电（java并行流底层原理）

在实际开发过程中，我们使用最多的程序结构是顺序结构，也就是序列 sequence，通常还会与循环结构的代码相混合 Loops顺序结构只能依次执行任务，消耗的时间是各个顺序任务的总和当对接口的请求时间有要求，需要对任务的处理时间进行压缩，转化到代码上就是对代码结构进行优化，我们希望将序列任务转化为并行任务 parallelism。

并行指的是同时发生，在计算机上为每个任务分配单独的 CPU 核心，同时进行任务处理完毕后进行汇总如果你了解 Hadoop ，这种方式与 MapReduce 如出一辙我们可以调用 parallelStream 方法来把集合转换为并行流。

并行流就是一个把内容分成多个数据块，并用不同的线程分别处理每个数据块的流这样就可以把工作负荷分配给多核处理器的所有内核，让它们都忙起来什么是流流是 JavaAPI 的新成员，它允许你以声明性方式处理数据集合（通过查询语句来表达，而不是临时编写一个实现）。

就现在来说，你可以把它们看成遍历数据集的高级迭代器Stream支持顺序 sequential 和并行 parallel 聚合操作的元素序列下面的示例演示了使用 Stream和 IntStream 的聚合操作。

int sum = widgets.stream().filter(w -> w.getColor()== RED).mapToInt(w -> w.getWeight()).sum();在这个例子中，widgets是一个 Collection

我们通过以下方法创建一个小部件对象流 Collection.stream()该方法返回以此集合为源的顺序流如上方法过滤颜色为红色的 Widgets 并将 Collection转化为Collection的流，并对所有int流求和。

流在与集合主要不同：没有存储空间流不是存储元素的数据结构本质上是功能性的对流的操作会产生结果，但不会修改其源例如，上面的例子不会修改 Collection消耗品在流的生存期内，流中元素只能访问一次流操作分为中间（intermediate ）操作和终端（terminal ）操作，并合并以形成流管道（stream pipelines）。

流管道由源（例如 Collection，数组，生成器函数或I / O通道）组成；随后是零个或多个中间操作，例如 Stream.filter 或 Stream.map；以及诸如Stream.forEach 或终端操作 Stream.reduce。

终端操作（例如Stream.forEach或 IntStream.sum）可能会遍历流执行操作执行终端操作后，流管道被视为已消耗，无法再使用；如果需要再次遍历同一数据源，则必须返回到数据源以获取新的流流的各种操作

我们使用一个 Menu 菜单列表的例子来看一下流的各种操作List menu = Arrays.asList( new Dish("pork", false,800, Dish.Type.MEAT), new Dish("beef", false,700, Dish.Type.MEAT), new Dish("chicken", false,400, Dish.Type.MEAT), new Dish("french fries", true,530, Dish.Type.OTHER), new Dish("rice", true,350, Dish.Type.OTHER), new Dish("season fruit", true,120, Dish.Type.OTHER), new Dish("pizza", true,550, Dish.Type.OTHER), new Dish("prawns", false,300, Dish.Type.FISH), new Dish("salmon", false,450, Dish.Type.FISH));public class Dish { private final String name; private final boolean vegetarian; private final int calories; private final Type type; public boolean isVegetarian(){ return vegetarian;} public enum Type { MEAT, FISH, OTHER }//...省略 Getter Setter}

如上的集合可以使用如下流操作import static java.util.Comparator.comparing; import static java.util.stream.Collectors.toList; List

lowCaloricDishesName = menu.stream().filter(d -> d.getCalories()<400)//选出400卡路里以下的菜肴.sorted(comparing(Dish::getCalories))//按照卡路里排序.map(Dish::getName)//提取菜肴的名称.limit(3)//只选择头三个.collect(toList());//将所有名称保存到 List 中

为了 利用 多核 架构 并行 执行 这段 代码，你 只需 要把 stream()换 成 parallelStream()：List lowCaloricDishesName = menu.parallelStream().filter(d -> d.getCalories()<400).sorted(comparing( Dishes::getCalories)).map(Dish::getName).limit(3).collect(toList());

数据处理操作——流的数据处理功能支持类似于数据库的操作，以及函数式编程语言中的常用操作，如 filter 、map 、reduce 、find 、match 、sort 等流操作可以顺序执行，也可并行执行。

如上操作中filter ——接受 Lambda，从流中排除某些信息map ——接受 Lambada，将元素转化为其他形式或提取信息limit ——截断流，使元素不超过指定数量collect ——将流转化为其他形式。

你可以看到两类操作：filter、map 和 limit 可以连成一条流水线；collect 触发流水线执行并关闭它可以连接起来的流操作称为中间操作，关闭流的操作称为终端操作流的三要素一个数据源（如集合）来执行一个查询；一个中间操作链，形成一条流的流水线；一个终端操作，执行流水线，并能生成结果。

中间操作：filter、map、limit、sorted、distinct终端操作：forEach、count、collect流的中间操作使用谓词筛选filter()操作会接受一个谓词（一个返回boolean的函数）作为参数，并返回一个包括所有符合谓词的元素的流。

filter(d -> d.getCalories()<400)这个代码中定义了一个匿名内部类，这个用 Lambda 表达式定义的方法返回 d.getCalories()<400这个表达式的值，随后 filter 会根据过滤条件进行流过滤。

也可以使用 filter(Dish::isVegetarian)，有双冒号的用法，就是把方法当做参数传到stream内部，使 stream 的每个元素都传入到该方法里面执行一下Dish::isVegetarian 返回每个元素的 true 或 false 值。

映射map 方法会接受一个函数作为参数这个函数会被应用到每个元素上，并将其映射成一个新的元素List dishNames = menu.stream().map(Dish::getName).collect(toList());。

map 方法返回的流类型是 Stream匹配anyMatch 方法返回 Boolean 值，用来查找流中是否包含某个值，是一个终端操作if( menu.stream().anyMatch(Dish::isVegetarian)){ System.out.println(" The menu is vegetarian friendly!!");}。

AllMatch 方法也返回 Boolean 值，用来判断是否流中元素是否全部匹配NoneMatch 方法确保流中没有任何元素与谓词相匹配查找可以使用流的 find 方法查找流中元素，可以使用顺序查找 findFirst()或任意查找 findAny()。

查找方法是终端操作List someNumbers = Arrays.asList(1,2,3,4,5); Optional firstSquareDivisibleByThree = someNumbers.stream().map( x -> x * x).filter( x -> x %3 == 0).findFirst();//9

规约操作规约操作指将流规约成一个值也可称为折叠 fold元素求和int sum =0; for (int x : numbers){ sum += x;}//使用流的写法int sum = numbers.stream().reduce(0,(a, b)-> a + b);。

reduce 接受两个参数：一个初始值，这里是0；一个 BinaryOperator 来将两个元素结合起来产生一个新值，这里我们用的是 lambda (a,b)->a+b如上的例子中，0作为 a 的第一个参数，从流中获取 numbers 列表中的一个数值如3 作为 b ，两者之和3 作为 a 的第二个参数与 numbers 列表获取的第二个数值 b =4相加，直到所有 numbers 列表中数据都参与计算返回最终的 a + b 的 sum 值。

最大值与最小值Optional max = numbers.stream().reduce(Integer::max);Optional min = numbers.stream().reduce(Integer::min);//也可以使用 Lambda (X,Y)->X

数值流int calories = menu.stream().map(Dish:: getCalories).reduce(0, Integer:: sum);int calories = menu.stream()//返回一个 Stream

.mapToInt(Dish::getCalories)//返回一个 IntStream.sum();IntStream intStream = menu.stream(). mapToInt( Dish:: getCalories);//将 Stream 转换为数值流 Stream

stream = intStream.boxed();//将数值流转换为 Stream并行流并行计算数值和public static long parallelSum( long n){ return Stream.iterate(1L, i -> i +1).limit(n).parallel()//将流转换为并行流.reduce(0L, Long:: sum);}

并行流默认的线程数量是你的处理器数量，要保证在内核中并行执行工作的时间比在内核之间传输数据的时间长使用并行流要避免共享对象可变状态，并且要避免自动拆装箱对于较小的数据一般不使用并行流，并行处理少数几个元素可能耗费时间增多，因为移动多个核的数据代价也很大。

参考资料《 Java 8实战》 .人民邮电出版社.JDK 11官方文档