java8新特性之数据流（串行流和并行流）

本文主要是介绍java8新特性之数据流（串行流和并行流），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1.什么是数据流（what）

java.util.Stream表示了某一种元素的序列，在这些元素上可以进行各种操作。

2.解决什么问题（why）

Stream操作可以是中间操作，也可以是完结操作。完结操作会返回一个某种类型的
值，而中间操作会返回流对象本身，并且你可以通过多次调用同一个流操作方法来将操作结果串起来！

3.怎么用（waht do）

3.1 语法

注：

中间操作：返回的结果是还是Stream流，允许在其返回结果上继续进行流的操作
中止操作：不返回流，所以我们不能再调用其他的流操作
所有的demo都会在3.2
（大家生理
1. Filter（中间操作）
接受一个Boolean类型的变量，并将所有流对象中的元素进行过滤
3. Sorted
Sorted是一个中间操作，能够返回一个排过序的流对象的视图。可指定排序规则!默认正序，逆序：orted(Comparator.reverseOrder())（注意：他的排序是按照ASCII码值排序）
4. Map
map是一个对于流对象的中间操作，通过给定的方法，它能够把流对象中的每一个元素对应到另外一个对象上。
5. Match
所有的匹配操作都是终结操作。匹配操作有多种不同的类型，都是用来判断某一种规则是否与流对象相互吻合的。只返回一个boolean类型的结果。有anyMatch、allMatch、noneMatch。
6. Count
Count是一个终结操作，它的作用是返回一个数值，用来标识当前流对象中包含的元素数量。
7. Reduce
Reduce操作是一个终结操作，它能够通过某一个方法，对元素进行削减操作。该操作的结果会放在一个Optional变量里返回。
8. ForEach
ForEach是一个中止操作。ForEach接受一个function接口类型的变量，用来执行对每一个元素的操作。。它不返回流，所以我们不能再调用其他的流操作。

3.2 示例

//通过串行流实现过滤判断。List<String> strings = Arrays.asList("abc", "", "bc", "efg", "abed", "", "jkl");//Tips limit获取指定数量的流List<String> filtered = strings.stream().limit(3).filter(string -> !string.isEmpty()).collect(Collectors.toList());filtered.forEach(System.out::println);System.out.println("====================");//通过并行流实现迭代List<String> parallelString = strings.parallelStream().filter(s -> !s.isEmpty()).collect(Collectors.toList());parallelString.forEach(System.out::println);//forEach来迭代流中的每个数据System.out.println("forEach来迭代流中的每个数据");parallelString.forEach(s->{s="er";System.out.println(s);});//map 方法用于映射每个元素到对应的结果，也可以用于迭代流中的元素List<Integer> numbers = Arrays.asList(3, 2, 2, 3, 7, 3, 5);//Tips: distinct:去重作用List<Integer> squaresList = numbers.stream().map(i -> i * i).distinct().collect(Collectors.toList());squaresList.forEach(System.out::println);List<String> stringList = parallelString.stream().map(s -> s + s).distinct().collect(Collectors.toList());stringList.forEach(System.out::println);//filter 方法用于通过设置的条件过滤出元素。以下代码片段使用 filter 方法过滤出空字符串：List<String> filterString = Arrays.asList("abc", "", "bc", "efg", "abed", "", "jkl");// 获取空字符串的数量long count = filterString.stream().filter(String::isEmpty).count();List<String> filterCollect = filterString.stream().filter(String::isEmpty).collect(Collectors.toList());System.out.println(count);System.out.println(filterCollect.size());//sorted 方法用于对流进行排序。以下代码片段使用 sorted 方法对输出的 10 个随机数进行排序!Random random = new Random();random.ints().limit(10).sorted().forEach(System.out::println);//Collectors 类实现了很多归约操作，例如将流转换成集合和聚合元素。Collectors 可用于返回列表或字符串List<String> collectList = strings.stream().filter(string -> !string.isEmpty()).collect(Collectors.toList());System.out.println("筛选列表: " + collectList);String mergedString = strings.stream().filter(string -> !string.isEmpty()).collect(Collectors.joining(""));System.out.println("合并字符串: " + mergedString);List<Integer> integerList = Arrays.asList(3, 2, 2, 3, 7, 3, 5);//一些产生统计结果的收集器也非常有用。它们主要用于int、double、long等基本类型上，它们可以用来产生类似如下的统计结果。IntSummaryStatistics stats = integerList.stream().mapToInt((x) -> x).summaryStatistics();System.out.println("列表中最大的数 : " + stats.getMax());System.out.println("列表中最小的数 : " + stats.getMin());System.out.println("所有数之和 : " + stats.getSum());System.out.println("平均数 : " + stats.getAverage());

4.并行流

4.1并行流操作

流操作可以是顺序的，也可以是并行的。顺序操作通过单线程执行，而并行操作则通过多线程执行。

package com.jimi.java8.streamapi;import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;/*** @author test*/
public class ParallelStreamAndStream {public static void main(String[] args) {List<Integer> i = Arrays.asList(1, 2, 3, 4, 5, 6, 7);System.out.println("stream.forEach打印结果如下(有序)");i.stream().forEach(System.out::println);System.out.println("parallelStream.forEach打印结果如下(无序)");i.parallelStream().forEach(System.out::println);System.out.println("parallelStream.forEachOrdered打印结果如下(有序.结果同stream.forEac)");i.parallelStream().forEachOrdered(System.out::println);//模拟10000条数据 循环打印测试List<Integer> list = new ArrayList();for (int j = 0; j < 1000; j++) {list.add(j);}// 统计并行执行list的线程Set<Thread> threadSet = new CopyOnWriteArraySet<>();// 并行执行list.parallelStream().forEach(integer -> {Thread thread = Thread.currentThread();// 统计并行执行list的线程threadSet.add(thread);});System.out.println(threadSet);//打印结果由此证明parallelStream是多管道线程System.out.println("threadSet一共有" + threadSet.size() + "个线程");System.out.println("系统一个有" + Runtime.getRuntime().availableProcessors() + "个cpu");list.parallelStream().forEach(System.out::println);}
}

4.2 并行流效率测试

package com.jimi.java8.streamapi;import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.TimeUnit;/*** @author test*/
public class ParallelTime {public static void main(String[] args) throws InterruptedException {//模拟10000条数据 forEach打印测试List<Integer> list = new ArrayList();for (int j = 0; j < 10; j++) {list.add(j);}long startTime = System.currentTimeMillis();//传统for循环for (int i = 0; i < list.size(); i++) {try {//睡眠1毫秒TimeUnit.MILLISECONDS.sleep(1);} catch (Exception e) {e.printStackTrace();}}long endTime = System.currentTimeMillis();System.out.println("传统for循环运行时间:" + (endTime - startTime) + "ms");// 测试单管道stream执行效率startTime = System.currentTimeMillis();list.stream().forEach(r -> {try {//睡眠1毫秒TimeUnit.MILLISECONDS.sleep(1);} catch (Exception e) {e.printStackTrace();}});long streamendTime = System.currentTimeMillis();System.out.println("stream : " + (streamendTime - startTime) + "ms");// 测试多管道parallelStream执行效率startTime = System.currentTimeMillis();Set<Thread>  threads = new CopyOnWriteArraySet<>();list.parallelStream().forEach(r -> {try {Thread thread = Thread.currentThread();threads.add(thread);System.out.println("系统创建了"+threads.size()+"个线程");//睡眠1毫秒TimeUnit.MILLISECONDS.sleep(1);} catch (Exception e) {e.printStackTrace();}});long parallelStreamendTime = System.currentTimeMillis();System.out.println("parallelStream : " + (parallelStreamendTime - startTime) + "ms");}
}

并行流快了很多。但是我产生了另外的一个问题便是：有时候在排查错误的时候，流的形式不方便进行日志的监控。
在这里插入图片描述