java8之Stream

1. Stream介绍

Stream不是集合元素，它不是数据结构并不保存数据，它是有关算法和计算的，它更像一个高级版本的Iterator。原始版本的Iterator，用户只能显式地一个一个遍历元素并对其执行某些操作；高级版本的Stream，用户只要给出需要对其包含的元素执行什么操作，比如，“过滤掉长度大于 10 的字符串”、“获取每个字符串的首字母”等，Stream会隐式地在内部进行遍历，做出相应的数据转换。Stream就如同一个迭代器（Iterator），单向，不可往复，数据只能遍历一次，遍历过一次后即用尽了。
在Java8 中增加了Stream API ，简化了串行或者并行的大批量的操作 。
Stream中的数据元素可以是对象引用，或者基本数据类型的值： int long double

2. Stream的具体用法

1. Stream 常用的创建方法

   ***使用Collection下的 stream() 和 parallelStream() 方法***　　

       
 List<String> list = new ArrayList<String>();

//Returns a sequential {@code Stream} with this collection as its source.(返回一个顺序的流，将此集合作为其源。)
  Stream<String> stream = list.stream(); 
 //Returns a possibly parallel {@code Stream} with this collection as its (返回一个可能是并行的流，其中包含此集合)
  Stream<String> parallelStream = list.parallelStream();

• 使用Arrays 中的 stream() 方法，将数组转成流

Integer[] nums = new Integer[10];
Stream<Integer> stream = Arrays.stream(nums);

• 使用Stream中的静态方法：of()、iterate()、generate()

Stream<Integer> stream = Stream.of(1,2,3,4,5,6);

Stream<Integer> stream2 = Stream.iterate(0, (x) -> x + 2).limit(6);
stream2.forEach(System.out::println); // 0 2 4 6 8 10

Stream<Double> stream3 = Stream.generate(Math::random).limit(2);
stream3.forEach(System.out::println);

• 使用 BufferedReader.lines() 方法，将每行内容转成流

  BufferedReader reader = new BufferedReader(new FileReader("D:\\test_stream.txt"));
  Stream<String> lineStream = reader.lines();
  lineStream.forEach(System.out::println);

• 使用 Pattern.splitAsStream() 方法，将字符串分隔成流

Pattern pattern = Pattern.compile(",");
Stream<String> stringStream = pattern.splitAsStream("a,b,c,d");
stringStream.forEach(System.out::println);

2 . Stream的中间操作

• 筛选与切片

​ filter：过滤流中的某些元素
　　 limit(n)：获取n个元素
　　 skip(n)：跳过n个元素，配合limit(n)可实现分页
　　 distinct：通过流中元素的 hashCode() 和 equals() 去除重复元素

Stream<Integer> stream = Stream.of(6, 4, 6, 7, 3, 9, 8, 10, 12, 14, 14);
  
Stream<Integer> newStream = stream.filter(s -> s > 5) //6 6 7 9 8 10 12 14 14
.distinct() //6 7 9 8 10 12 14
.skip(2) //9 8 10 12 14
.limit(2); //9 8
newStream.forEach(System.out::println);

List<Student> stuList = stuList();
 
	// 需要筛选的条件：从stuList中筛选出年龄为21和22的学生
	List<Integer> ageList = new ArrayList<Integer>();
	ageList.add(21);
	ageList.add(22);
 
	// JDK1.8提供了lambda表达式， 可以从stuList中过滤出符合条件的结果。
	// 定义结果集
	List<Student> result = null;
	result = stuList.stream()
			.filter((Student s) -> ageList.contains(s.getAge()))
			.collect(Collectors.toList());
   
	// 打印原有stuList集合中的数据
	System.out.println("原有stuList集合中的数据");
	stuList.forEach((Student s) -> System.out.println(s.getName() + "--->" + s.getAge()));
   

• 映射

​ map：接收一个函数作为参数，该函数会被应用到每个元素上，并将其映射成一个新的元素。
　 flatMap：接收一个函数作为参数，将流中的每个值都换成另一个流，然后把所有流连接成一个流。

List<String> list = Arrays.asList("a,b,c", "1,2,3");
  
//将每个元素转成一个新的且不带逗号的元素
Stream<String> s1 = list.stream().map(s -> s.replaceAll(",", ""));
s1.forEach(System.out::println); // abc 123
  
Stream<String> s3 = list.stream().flatMap(s -> {
//将每个元素转换成一个stream
String[] split = s.split(",");
Stream<String> s2 = Arrays.stream(split);
return s2;
});
s3.forEach(System.out::println); // a b c 1 2 3

• 排序

​ sorted()：自然排序，流中元素需实现Comparable接口
　　sorted(Comparator com)：定制排序，自定义Comparator排序器

List<String> list = Arrays.asList("aa", "ff", "dd");
//String 类自身已实现Compareable接口
list.stream().sorted().forEach(System.out::println);// aa dd ff
  
Student s1 = new Student("aa", 10);
Student s2 = new Student("bb", 20);
Student s3 = new Student("aa", 30);
Student s4 = new Student("dd", 40);
List<Student> studentList = Arrays.asList(s1, s2, s3, s4);
  
//自定义排序：先按姓名升序，姓名相同则按年龄升序
studentList.stream().sorted(
(o1, o2) -> {
if (o1.getName().equals(o2.getName())) {
return o1.getAge() - o2.getAge();
} else {
return o1.getName().compareTo(o2.getName());
}
}
).forEach(System.out::println);　

• 消费

　　peek：如同于map，能得到流中的每一个元素。但map接收的是一个Function表达式，有返回值；而peek接收的是Consumer表达 式，没有返回值。

Student s1 = new Student("aa", 10);
Student s2 = new Student("bb", 20);
List<Student> studentList = Arrays.asList(s1, s2);
  
studentList.stream()
.peek(o -> o.setAge(100))
.forEach(System.out::println);
  
//结果：
Student{name='aa', age=100}
Student{name='bb', age=100}

3. Stream的终止操作 (常用函数)

• 匹配 聚合操作

List<Integer> list = Arrays.asList(1, 2, 3, 4, 5);
  
boolean allMatch = list.stream().allMatch(e -> e > 10); //false
boolean noneMatch = list.stream().noneMatch(e -> e > 10); //true
boolean anyMatch = list.stream().anyMatch(e -> e > 4); //true
  
Integer findFirst = list.stream().findFirst().get(); //1
Integer findAny = list.stream().findAny().get(); //1
  
long count = list.stream().count(); //5
Integer max = list.stream().max(Integer::compareTo).get(); //5
Integer min = list.stream().min(Integer::compareTo).get(); //1　

• 规约操作(reduce)

三个重载的方法：

1.Optional<T> reduce(BinaryOperator<T> accumulator);

2.T reduce(T identity, BinaryOperator<T> accumulator)

3.<U> U reduce(U identity, BiFunction<U, ? super T, U> accumulator, BinaryOperator<U> combiner);用在流并发操作的时候，将每个线程前两个参数形成的结果result集合并为一个。所以第三个参数是一个BinaryOperator函数接口

//reduce(String identity, BinaryOperator<String> accumulator) 第一个参数相当于起始值 ，第二个参数是二元运算函数接口
String s1 = custs.stream().map(Cust::getCustName).reduce("", (x,y)->x+","+y).replaceFirst(",", "");
System.out.println(s1);
		
//reduce(BinaryOperator<String> accumulator) 参数是二元运算函数接口，因为没有给默认值，所以为了避免使用时空指针异常，返回的是Optional
Optional<String> opt = custs.stream().map(Cust::getCustName).reduce((x,y)->x+","+y);
String s2 = opt.get().replaceFirst(",", "");

• 收集操作
  collect: 接收一个Collector实例，将流中元素收集成另外一个数据结构。
  Collector<T, A, R>: 是一个接口，有以下5个抽象方法：
  ***Supplier<A> supplier():***创建一个结果容器A
  ***BiConsumer<A, T> accumulator():***消费型接口，第一个参数为容器A，第二个参数为流中元素T。
  ***BinaryOperator<A> combiner():***函数接口，该参数的作用跟上一个方法(reduce)中的combiner 参数一样，将并行流中各 个子进程 的运行结果(accumulator函数操作后的容器A)进行合并。
  ***Function<A, R> finisher():***函数式接口，参数为：容器A，返回类型为：collect方法最终想要的结果R。
  Set<Characteristics> characteristics(): 返回一个不可变的Set集合，用来表明该Collector的特征。有以下三个特征：
  ***CONCURRENT:***表示此收集器支持并发。
  ***UNORDERED:***表示该收集操作不会保留流中元素原有的顺序。
  ***IDENTITY_FINISH:***表示finisher参数只是标识而已，可忽略。

Collector 工具库：Collectors

Student s1 = new Student("aa", 10,1);
Student s2 = new Student("bb", 20,2);
Student s3 = new Student("cc", 10,3);
List<Student> list = Arrays.asList(s1, s2, s3);

//装成list
List<Integer> ageList = list.stream().map(Student::getAge).collect(Collectors.toList()); // [10, 20, 10]

//转成set
Set<Integer> ageSet = list.stream().map(Student::getAge).collect(Collectors.toSet()); // [20, 10]

//转成map,注:key不能相同，否则报错
Map<String, Integer> studentMap = list.stream().collect(Collectors.toMap(Student::getName, Student::getAge)); // {cc=10, bb=20, aa=10}

//转成map 如何key相同 是否覆盖  (oldVal ,newVal)->newVal ? oldVal
 Map<String, Integer> studentMap = list.stream().collect(Collectors.toMap(Student::getName, Student::getAge, (oldVal ,newVal)->newVal)); 

//转为 LinkedHashMap   (LinkedHashMap::new)
LinkedHashMap<String, Integer> collect = list.stream().collect(Collectors.toMap(Student::getName, Student::getAge, (oldVal, newVal) -> newVal, LinkedHashMap::new));// {cc=10, bb=20, aa=10}


//字符串分隔符连接
String joinName = list.stream().map(Student::getName).collect(Collectors.joining(",", "(", ")")); // (aa,bb,cc)

//聚合操作
//1.学生总数
Long count = list.stream().collect(Collectors.counting()); // 3
//2.最大年龄 (最小的minBy同理)
Integer maxAge = list.stream().map(Student::getAge).collect(Collectors.maxBy(Integer::compare)).get(); // 20
//3.所有人的年龄
Integer sumAge = list.stream().collect(Collectors.summingInt(Student::getAge)); // 40
//4.平均年龄
Double averageAge = list.stream().collect(Collectors.averagingDouble(Student::getAge)); // 13.333333333333334
// 带上以上所有方法
DoubleSummaryStatistics statistics = list.stream().collect(Collectors.summarizingDouble(Student::getAge));
System.out.println("count:" + statistics.getCount() + ",max:" + statistics.getMax() + ",sum:" + statistics.getSum() + ",average:" + statistics.getAverage());

//分组
Map<Integer, List<Student>> ageMap = list.stream().collect(Collectors.groupingBy(Student::getAge));
//多重分组,先根据类型分再根据年龄分
Map<Integer, Map<Integer, List<Student>>> typeAgeMap = list.stream().collect(Collectors.groupingBy(Student::getType, Collectors.groupingBy(Student::getAge)));

//分区
//分成两部分，一部分大于10岁，一部分小于等于10岁
Map<Boolean, List<Student>> partMap = list.stream().collect(Collectors.partitioningBy(v -> v.getAge() > 10));

//规约
Integer allAge = list.stream().map(Student::getAge).collect(Collectors.reducing(Integer::sum)).get(); //40　

3.补充

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

​ 顺序流： 单线程执行

/**
 * parallel（）  开启并行流
 */
Instant start = Instant.now();
LongStream.rangeClosed(1,10000000000L).parallel().reduce(0,Long::sum);
System.out.println("系统执行时间:"+ Duration.between(start,Instant.now()).toMillis());

/**
 * sequential()  开启顺序流
 */
Instant start1 = Instant.now();
LongStream.rangeClosed(1,10000000000L).sequential().reduce(0,Long::sum);
System.out.println("系统执行时间:"+Duration.between(start1,Instant.now()).toMillis());

注意：

​ 并行流 ：

​ 使用并行流并不是一定会提高效率，因为jvm对数据进行切片和切换线程也是需要时间的。

​ 所以数据量越小，串行操作越快；数据量越大，并行操作效果越好。

​ 并行流内部使用了默认的ForkJoinPool线程池，所以它默认的线程数量就是处理器的数量，通过它可以得到这个值。
​ System.out.println(Runtime.getRuntime().availableProcessors())。

​ 通过这个方法可以修改这个值，而且这个还是全局属性，不过建议一般不修改

​ System.setProperty(“java.util.concurrent.ForkJoinPool.common.parallelism”, “12”);