重学JDK8新特性之Stream

集合处理数据的弊端

当我们在需要对集合中的元素进行操作的时候，除了必需的添加，删除，获取外，最典型的操作就是
集合遍历

public static void main(String[] args) {
	// 定义一个List集合
	List<String> list = Arrays.asList("张三","张三丰","成龙","周星驰");
	// 1.获取所有 姓张的信息
	List<String> list1 = new ArrayList<>();
	for (String s : list) {
		if(s.startsWith("张")){
			list1.add(s);
		}
	}
	// 2.获取名称长度为3的用户
	List<String> list2 = new ArrayList<>();
	for (String s : list1) {
		if(s.length() == 3){
			list2.add(s);
		}
	}
	// 3. 输出所有的用户信息
	for (String s : list2) {
		System.out.println(s);
	}
}

上面的代码总是一次次的循环，或许需要一种更加高效的处理方式，这时可以通过JDK8中提供的Stream API来解决这个问题。

public static void main(String[] args) {
	// 定义一个List集合
	List<String> list = Arrays.asList("张三","张三丰","成龙","周星驰");
	// 1.获取所有 姓张的信息
	// 2.获取名称长度为3的用户
	// 3. 输出所有的用户信息
	list.stream()
		.filter(s->s.startsWith("张"))
		.filter(s->s.length() == 3)
		.forEach(s->{
			System.out.println(s);
		});
	System.out.println("----------");
	list.stream()
		.filter(s->s.startsWith("张"))
		.filter(s->s.length() == 3)
		.forEach(System.out::println);
}

如果使用Stream来实现，会显得更加简便，并且可读性也很好。

Stream思想

注意：Stream和IO流(InputStream/OutputStream)没有任何关系，请暂时忘记对传统IO流的固有印象！
Stream流式思想类似于工厂车间的“生产流水线”，Stream流不是一种数据结构，不保存数据，而是对数据进行加工
处理。Stream可以看作是流水线上的一个工序。在流水线上，通过多个工序让一个原材料加工成一个商品。

Stream API能让我们快速完成许多复杂的操作，如筛选、切片、映射、查找、去除重复，统计，匹配和归约。

Stream流的获取方式

根据Collection获取

首先，java.util.Collection 接口中加入了default方法 stream，也就是说Collection接口下的所有的实
现都可以通过steam方法来获取Stream流。

public static void main(String[] args) {
	List<String> list = new ArrayList<>();
	list.stream();
	Set<String> set = new HashSet<>();
	set.stream();
	Vector vector = new Vector();
	vector.stream();
}

而Map比较特殊，如果想要使用，需要根据Map来获取对应的Key和Value的集合

public static void main(String[] args) {
	Map<String,Object> map = new HashMap<>();
	Stream<String> stream = map.keySet().stream(); // key
	Stream<Object> stream1 = map.values().stream(); // value
}

通过Stream的of方法

在实际开发中我们不可避免的还是会操作到数组中的数据，由于数组对象不可能添加默认方法，所有Stream接口中提供了静态方法of

public static void main(String[] args) {
	Stream<String> a1 = Stream.of("a1", "a2", "a3");
	String[] arr1 = {"aa","bb","cc"};
	Stream<String> arr11 = Stream.of(arr1);
	Integer[] arr2 = {1,2,3,4};
	Stream<Integer> arr21 = Stream.of(arr2);
	arr21.forEach(System.out::println);
	// 注意：基本数据类型的数组是不行的
	int[] arr3 = {1,2,3,4};
	Stream.of(arr3).forEach(System.out::println);
}

Stream常用方法介绍

Stream流模型的操作很丰富，这里介绍一些常用的API。这些方法可以被分成两种：

在这里插入图片描述

终结方法：返回值类型不再是 Stream 类型的方法，不再支持链式调用

非终结方法：返回值类型仍然是 Stream 类型的方法，支持链式调用

Stream注意事项(重要)

Stream只能操作一次
Stream方法返回的是新的流
Stream不调用终结方法，中间的操作不会执行

foreach

forEach用来遍历流中的数据的

void forEach(Consumer<? super T> action);

该方法接受一个Consumer接口，会将每一个流元素交给函数处理

public static void main(String[] args) {
	Stream.of("a1", "a2", "a3").forEach(System.out::println);;
}

count

Stream流中的count方法用来统计其中的元素个数的。该方法返回一个long值，代表元素的个数。

public static void main(String[] args) {
	long count = Stream.of("a1", "a2", "a3").count();
	System.out.println(count);
}

filter

filter方法的作用是用来过滤数据的。返回符合条件的数据

可以通过filter方法将一个流转换成另一个子集流

Stream<T> filter(Predicate<? super T> predicate);

该接口接收一个Predicate函数式接口参数作为筛选条件

public static void main(String[] args) {
	Stream.of("a1", "a2", "a3","bb","cc","aa","dd")
	.filter((s)->s.contains("a"))
	.forEach(System.out::println);
}

输出：

a1
a2
a3
aa

limit

limit方法可以对流进行截取处理，支取前n个数据

参数是一个long类型的数值，如果集合当前长度大于参数就进行截取，否则不操作：

public static void main(String[] args) {
	Stream.of("a1", "a2", "a3","bb","cc","aa","dd")
	.limit(3)
	.forEach(System.out::println);
}

输出：

a1
a2
a3

skip

如果希望跳过前面几个元素，可以使用skip方法获取一个截取之后的新流：

public static void main(String[] args) {
	Stream.of("a1", "a2", "a3","bb","cc","aa","dd")
	.skip(3)
	.forEach(System.out::println);
}

输出：

bb
cc
aa
dd

map

如果我们需要将流中的元素映射到另一个流中，可以使用map方法：

<R> Stream<R> map(Function<? super T, ? extends R> mapper);

该接口需要一个Function函数式接口参数，可以将当前流中的T类型数据转换为另一种R类型的数据

public static void main(String[] args) {
	Stream.of("1", "2", "3","4","5","6","7")
	//.map(msg->Integer.parseInt(msg))
	.map(Integer::parseInt)
	.forEach(System.out::println);
}

sorted

如果需要将数据排序，可以使用sorted方法：

在使用的时候可以根据自然规则排序，也可以通过比较强来指定对应的排序规则

public static void main(String[] args) {
	Stream.of("1", "3", "2","4","0","9","7")
	//.map(msg->Integer.parseInt(msg))
	.map(Integer::parseInt)
	//.sorted() // 根据数据的自然顺序排序
	.sorted((o1,o2)->o2-o1) // 根据比较强指定排序规则
	.forEach(System.out::println);
}

distinct

如果要去掉重复数据，可以使用distinct方法：

public static void main(String[] args) {
	Stream.of("1", "3", "3","4","0","1","7")
	//.map(msg->Integer.parseInt(msg))
	.map(Integer::parseInt)
	//.sorted() // 根据数据的自然顺序排序
	.sorted((o1,o2)->o2-o1) // 根据比较强指定排序规则
	.distinct() // 去掉重复的记录
	.forEach(System.out::println);
	System.out.println("--------");
	Stream.of(
	new Person("张三",18)
	,new Person("李四",22)
	,new Person("张三",18)
	).distinct()
	.forEach(System.out::println);
}

Stream流中的distinct方法对于基本数据类型是可以直接出重的，但是对于自定义类型，我们是需要
重写hashCode和equals方法来移除重复元素。

match

如果需要判断数据是否匹配指定的条件，可以使用match相关的方法

boolean anyMatch(Predicate<? super T> predicate); // 元素是否有任意一个满足条件
boolean allMatch(Predicate<? super T> predicate); // 元素是否都满足条件
boolean noneMatch(Predicate<? super T> predicate); // 元素是否都不满足条件

public static void main(String[] args) {
	boolean b = Stream.of("1", "3", "3", "4", "5", "1", "7")
	.map(Integer::parseInt)
	//.allMatch(s -> s > 0)
	//.anyMatch(s -> s >4)
	.noneMatch(s -> s > 4);
	System.out.println(b);
}

注意match是一个终结方法

find

如果我们需要找到某些数据，可以使用find方法来实现

public static void main(String[] args) {
	Optional<String> first = Stream.of("1", "3", "3", "4", "5", "1","7").findFirst();
	System.out.println(first.get());
	Optional<String> any = Stream.of("1", "3", "3", "4", "5", "1","7").findAny();
	System.out.println(any.get());
}

max和min

如果我们想要获取最大值和最小值，那么可以使用max和min方法

Optional<T> min(Comparator<? super T> comparator);
Optional<T> max(Comparator<? super T> comparator);

public static void main(String[] args) {
	Optional<Integer> max = Stream.of("1", "3", "3", "4", "5", "1", "7")
	.map(Integer::parseInt)
	.max((o1,o2)->o1-o2);
	System.out.println(max.get());
	Optional<Integer> min = Stream.of("1", "3", "3", "4", "5", "1", "7")
	.map(Integer::parseInt)
	.min((o1,o2)->o1-o2);
	System.out.println(min.get());
}

终结方法

concat

如果有两个流，希望合并成为一个流，那么可以使用Stream接口的静态方法concat

public static void main(String[] args) {
	Stream<String> stream1 = Stream.of("a","b","c");
	Stream<String> stream2 = Stream.of("x", "y", "z");
	// 通过concat方法将两个流合并为一个新的流
	Stream.concat(stream1,stream2).forEach(System.out::println);
}

Stream结果收集

结果收集到集合

List<String> list = Stream.of("aa", "bb", "cc","aa")
	.collect(Collectors.toList());
System.out.println(list);
// 收集到 Set集合中
Set<String> set = Stream.of("aa", "bb", "cc", "aa")
	.collect(Collectors.toSet());
System.out.println(set);
// 如果需要获取的类型为具体的实现，比如：ArrayList HashSet
ArrayList<String> arrayList = Stream.of("aa", "bb", "cc", "aa")
//.collect(Collectors.toCollection(() -> new ArrayList<>()));
	.collect(Collectors.toCollection(ArrayList::new));
System.out.println(arrayList);
HashSet<String> hashSet = Stream.of("aa", "bb", "cc", "aa")
	.collect(Collectors.toCollection(HashSet::new));
System.out.println(hashSet);

输出：

[aa, bb, cc, aa]
[aa, bb, cc]
[aa, bb, cc, aa]
[aa, bb, cc]

结果收集到数组中

Object[] objects = Stream.of("aa", "bb", "cc", "aa")
	.toArray(); // 返回的数组中的元素是 Object类型
System.out.println(Arrays.toString(objects));
// 如果我们需要指定返回的数组中的元素类型
String[] strings = Stream.of("aa", "bb", "cc", "aa")
	.toArray(String[]::new);
System.out.println(Arrays.toString(strings));

对流中数据做聚合计算

当我们使用Stream流处理数据后，可以像数据库的聚合函数一样对某个字段进行操作，比如获得最大
值，最小值，求和，平均值，统计数量。

Optional<Person> maxAge = Stream.of(
		new Person("张三", 18)
		, new Person("李四", 22)
		, new Person("张三", 13)
		, new Person("王五", 15)
		, new Person("张三", 19)
).collect(Collectors.maxBy((p1, p2) -> p1.getAge() - p2.getAge()));
System.out.println("最大年龄：" + maxAge.get());
// 获取年龄的最小值
Optional<Person> minAge = Stream.of(
		new Person("张三", 18)
		, new Person("李四", 22)
		, new Person("张三", 13)
		, new Person("王五", 15)
		, new Person("张三", 19)
).collect(Collectors.minBy((p1, p2) -> p1.getAge() - p2.getAge()));
System.out.println("最新年龄:" + minAge.get());
// 求所有人的年龄之和
Integer sumAge = Stream.of(
		new Person("张三", 18)
		, new Person("李四", 22)
		, new Person("张三", 13)
		, new Person("王五", 15)
		, new Person("张三", 19)
)
//.collect(Collectors.summingInt(s -> s.getAge()))
.collect(Collectors.summingInt(Person::getAge))
;
System.out.println("年龄总和：" + sumAge);
// 年龄的平均值
Double avgAge = Stream.of(
		new Person("张三", 18)
		, new Person("李四", 22)
		, new Person("张三", 13)
		, new Person("王五", 15)
		, new Person("张三", 19)
).collect(Collectors.averagingInt(Person::getAge));
System.out.println("年龄的平均值：" + avgAge);
// 统计数量
Long count = Stream.of(
		new Person("张三", 18)
		, new Person("李四", 22)
		, new Person("张三", 13)
		, new Person("王五", 15)
		, new Person("张三", 19)
).filter(p->p.getAge() > 18)
.collect(Collectors.counting());
System.out.println("满足条件的记录数:" + count);

对流中数据做分组操作

// 根据账号对数据进行分组
Map<String, List<Person>> map1 = Stream.of(
		new Person("张三", 18, 175)
		, new Person("李四", 22, 177)
		, new Person("张三", 14, 165)
		, new Person("李四", 15, 166)
		, new Person("张三", 19, 182)
).collect(Collectors.groupingBy(Person::getName));
map1.forEach((k,v)-> System.out.println("k=" + k +"\t"+ "v=" + v));
System.out.println("-----------");
// 根据年龄分组 如果大于等于18 成年否则未成年
Map<String, List<Person>> map2 = Stream.of(
		new Person("张三", 18, 175)
		, new Person("李四", 22, 177)
		, new Person("张三", 14, 165)
		, new Person("李四", 15, 166)
		, new Person("张三", 19, 182)
).collect(Collectors.groupingBy(p -> p.getAge() >= 18 ? "成年" : "未成
年"));
map2.forEach((k,v)-> System.out.println("k=" + k +"\t"+ "v=" + v));

结果：

k=李四 v=[Person{name='李四', age=22, height=177}, Person{name='李四', age=15,
height=166}]
k=张三 v=[Person{name='张三', age=18, height=175}, Person{name='张三', age=14,
height=165}, Person{name='张三', age=19, height=182}]
-----------
k=未成年 v=[Person{name='张三', age=14, height=165}, Person{name='李四', age=15,
height=166}]
k=成年 v=[Person{name='张三', age=18, height=175}, Person{name='李四', age=22,
height=177}, Person{name='张三', age=19, height=182}]

对流中的数据做分区操作

Map<Boolean, List<Person>> map = Stream.of(
		new Person("张三", 18, 175)
		, new Person("李四", 22, 177)
		, new Person("张三", 14, 165)
		, new Person("李四", 15, 166)
		, new Person("张三", 19, 182)
).collect(Collectors.partitioningBy(p -> p.getAge() > 18));
map.forEach((k,v)-> System.out.println(k+"\t" + v));

结果：

false [Person{name='张三', age=18, height=175}, Person{name='张三', age=14,
height=165}, Person{name='李四', age=15, height=166}]
true [Person{name='李四', age=22, height=177}, Person{name='张三', age=19,
height=182}]

对流中的数据做拼接

String s1 = Stream.of(
		new Person("张三", 18, 175)
		, new Person("李四", 22, 177)
		, new Person("张三", 14, 165)
		, new Person("李四", 15, 166)
		, new Person("张三", 19, 182)
).map(Person::getName)
.collect(Collectors.joining());
// 张三李四张三李四张三
System.out.println(s1);
String s2 = Stream.of(
		new Person("张三", 18, 175)
		, new Person("李四", 22, 177)
		, new Person("张三", 14, 165)
		, new Person("李四", 15, 166)
		, new Person("张三", 19, 182)
).map(Person::getName)
.collect(Collectors.joining("_"));
// 张三_李四_张三_李四_张三
System.out.println(s2);

并行的Stream流

串行Stream流

前面使用的Stream流都是串行，也就是在一个线程上面执行。

Stream.of(5,6,8,3,1,6)
		.filter(s->{
		System.out.println(Thread.currentThread() + "" + s);
		return s > 3;
	}).count();

输出：

Thread[main,5,main]5
Thread[main,5,main]6
Thread[main,5,main]8
Thread[main,5,main]3
Thread[main,5,main]1
Thread[main,5,main]6

并行流

parallelStream其实就是一个并行执行的流，它通过默认的ForkJoinPool，可以提高多线程任务的速度。

我们可以通过两种方式来获取并行流。

通过List接口中的parallelStream方法来获取
通过已有的串行流转换为并行流(parallel)

List<Integer> list = new ArrayList<>();
	// 通过List 接口 直接获取并行流
	Stream<Integer> integerStream = list.parallelStream();
	// 将已有的串行流转换为并行流
	Stream<Integer> parallel = Stream.of(1, 2, 3).parallel();

并行操作

Stream.of(1,4,2,6,1,5,9)
		.parallel() // 将流转换为并发流，Stream处理的时候就会通过多线程处理
		.filter(s->{
		System.out.println(Thread.currentThread() + " s=" +s);
		return s > 2;
	}).count();

输出：

Thread[main,5,main] s=1
Thread[ForkJoinPool.commonPool-worker-2,5,main] s=9
Thread[ForkJoinPool.commonPool-worker-6,5,main] s=6
Thread[ForkJoinPool.commonPool-worker-13,5,main] s=2
Thread[ForkJoinPool.commonPool-worker-9,5,main] s=4
Thread[ForkJoinPool.commonPool-worker-4,5,main] s=5
Thread[ForkJoinPool.commonPool-worker-11,5,main] s=1

并行流和串行流对比

/**
* 普通for循环 消耗时间：138
*/
@Test
public void test01(){
	System.out.println("普通for循环:");
	long res = 0;
	for (int i = 0; i < times; i++) {
		res += i;
	}
}
/**
* 串行流处理
* 消耗时间：203
*/
@Test
public void test02(){
	System.out.println("串行流：serialStream");
	LongStream.rangeClosed(0,times)
		.reduce(0,Long::sum);
}
/**
* 并行流处理 消耗时间：84
*/
@Test
public void test03(){
	LongStream.rangeClosed(0,times)
	.parallel()
	.reduce(0,Long::sum);
}

通过案例我们可以看到parallelStream的效率是最高的。
Stream并行处理的过程会分而治之，也就是将一个大的任务切分成了多个小任务，这表示每个任务都是一个线程操作。

线程安全问题

在多线程的处理下，肯定会出现数据安全问题。如下：

List<Integer> list = new ArrayList<>();
for (int i = 0; i < 1000; i++) {
	list.add(i);
}
System.out.println(list.size());
List<Integer> listNew = new ArrayList<>();
// 使用并行流来向集合中添加数据
list.parallelStream()
		//.forEach(s->listNew.add(s));
		.forEach(listNew::add);
System.out.println(listNew.size());

实际上有可能出现线程安全问题：

java.lang.ArrayIndexOutOfBoundsException
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at
sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorI
mpl.java:62)
at
sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorA
ccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
at
java.util.concurrent.ForkJoinTask.getThrowableException(ForkJoinTask.java:598)
....
Caused by: java.lang.ArrayIndexOutOfBoundsException: 366
at java.util.ArrayList.add(ArrayList.java:463)

在并行流中往列表（List）集合进行添加元素可能会遇到线程安全问题。当多个线程同时执行添加操作时，由于并行流的特性，多个线程会同时访问和修改同一个列表对象，这可能导致以下问题：

竞态条件：由于多个线程同时修改列表，可能导致数据不一致或意外的结果。例如，如果两个线程同时向同一个索引位置添加元素，可能会导致其中一个元素被覆盖或丢失。
并发修改异常：Java 的 ArrayList 类并不是线程安全的，当多个线程同时修改列表时，可能会引发 ConcurrentModificationException 异常。

针对这个问题，我们的解决方案有哪些呢？

加同步锁
使用线程安全的容器
通过Stream中的toArray/collect操作

实现：

/**
* 加同步锁
*/
List<Integer> listNew = new ArrayList<>();
Object obj = new Object();
IntStream.rangeClosed(1,1000)
		.parallel()
		.forEach(i->{
			synchronized (obj){
				listNew.add(i);
			}
		});
System.out.println(listNew.size());

/**
* 使用线程安全的容器
*/

Vector v = new Vector();
Object obj = new Object();
IntStream.rangeClosed(1,1000)
		.parallel()
		.forEach(i->{
			synchronized (obj){
				v.add(i);
			}
		});
System.out.println(v.size());

/**
* 将线程不安全的容器转换为线程安全的容器
*/
List<Integer> listNew = new ArrayList<>();
// 将线程不安全的容器包装为线程安全的容器
List<Integer> synchronizedList = Collections.synchronizedList(listNew);
Object obj = new Object();
IntStream.rangeClosed(1,1000)
		.parallel()
		.forEach(i->{
			synchronizedList.add(i);
		});
System.out.println(synchronizedList.size());