java的stream的使用

最近发现开发中经常使用到streamAPI ，所有抽出一点时间，总结一些streamAPI ，一方面作为一个笔记，可以在开发不需要再去搜索，另一方面也希望加深对这个的认识

1.stream流式思想概述

stream流式思想类似工厂的流水线，在工厂中，原材料经过很多步工序，每步工序实现一个功能，最终得到我们需要的产品。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-RMlmn67u-1660866950464)(https://tse1-mm.cn.bing.net/th/id/OIP-C.A9go0od39qPlXqzbRCgyRwHaDg?pid=ImgDet&rs=1)]

2.stream的常用方法

方法名	方法作用	返回值类型	方法种类
count	统计个数	long	终结
forEach	逐一处理	void	终结
filter	过滤	Stream	函数拼接
limit	取用前几个	Stream	函数拼接
skip	跳过前几个	Stream	函数拼接
map	映射	Stream	函数拼接
concat	组合	Stream	函数拼接

终结方法：返回值类型不再是 Stream 类型的方法，不再支持链式调用。本小节中，终结方法包括 count 和 forEach 方法。

非终结方法：返回值类型仍然是 Stream 类型的方法，支持链式调用。（除了终结方法外，其余方法均为非终结方法。）

2.1代码中对象准备

package com.mashibing.stream.model;import lombok.Builder;
import lombok.Data;
import lombok.ToString;/*** @author fangdy* @date 2022-08-18 22:14*/
@Data
@Builder
@ToString
public class Person {/*** 姓名*/private String name;/*** 年龄*/private Integer age;/*** 性别*/private String gender;/*** 职业*/private String profession;/*** 身份证号码*/private String licenseNumber;
}

package com.mashibing.stream;import cn.hutool.core.collection.CollUtil;
import com.mashibing.stream.model.Person;
import org.junit.Test;import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;/*** @author fangdy* @date 2022-08-18 22:09*/
public class StreamTest {private static List<Person> personList = CollUtil.newArrayList(Person.builder().name("张三1").age(23).gender("男").profession("品质工程师").licenseNumber("110101199603074454").build(),Person.builder().name("张三1").age(26).gender("男").profession("品质工程师").licenseNumber("110101199603074454").build(),Person.builder().name("张三2").age(32).gender("男").profession("java工程师").licenseNumber("110101199603072838").build(),Person.builder().name("张三3").age(14).gender("男").profession("python工程师").licenseNumber("110101199603074315").build(),Person.builder().name("张三4").age(23).gender("男").profession("前端工程师").licenseNumber("110101199603078375").build(),Person.builder().name("张宁5").age(23).gender("女").profession("品质工程师").licenseNumber("110101199003073407").build(),Person.builder().name("张宁6").age(21).gender("女").profession("品质工程师").licenseNumber("110101199003073087").build(),Person.builder().name("张宁7").age(30).gender("女").profession("java工程师").licenseNumber("110101199003073861").build(),Person.builder().name("张宁8").age(18).gender("女").profession("python工程师").licenseNumber("110101199003070767").build(),Person.builder().name("张宁9").age(27).gender("女").profession("前端工程师").licenseNumber("110101199003071946").build());
}

2.2 filter,limit,skip,sorted,distinct, match, find,max和min,reduce方法,concat等对单对象的操作

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

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

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

Stream<T> limit(long maxSize);

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

Stream<T> skip(long n);

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

Stream<T> sorted();

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

Stream<T> distinct();

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

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

注意match是一个终结方法

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

 Optional<T> findFirst();Optional<T> findAny();

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

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

如果需要将所有数据归纳得到一个数据，可以使用reduce方法

T reduce(T identity, BinaryOperator<T> accumulator);

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

public static <T> Stream<T> concat(Stream<? extends T> a, Stream<? extends T> b) {Objects.requireNonNull(a);Objects.requireNonNull(b);@SuppressWarnings("unchecked")Spliterator<T> split = new Streams.ConcatSpliterator.OfRef<>((Spliterator<T>) a.spliterator(), (Spliterator<T>) b.spliterator());Stream<T> stream = StreamSupport.stream(split, a.isParallel() || b.isParallel());return stream.onClose(Streams.composedClose(a, b));}

   @Testpublic void singleObjTWo() {//  filter,limit,skip,sorted,distinct, match, find,max和min,reduce方法等对单对象的操作Stream.of("1", "2", "3","4","4","5").filter(x->!x.equals("1")).limit(4).skip(1).distinct().forEach(System.out::println);boolean b = Stream.of("1", "2", "3", "4", "4", "5").anyMatch(x -> x.equals("6"));System.out.println(b);String s = Stream.of("1", "2", "3", "4", "4", "5").findFirst().get();System.out.println(s);Integer integer = Stream.of("1", "2", "3", "4", "4", "5").map(Integer::parseInt).max((o1, o2) -> {return o1 - o2;}).get();System.out.println(integer);Integer sum = Stream.of(4, 5, 3, 9)// identity默认值// 第一次的时候会将默认值赋值给x// 之后每次会将 上一次的操作结果赋值给x y就是每次从数据中获取的元素.reduce(0, (x, y) -> {System.out.println("x="+x+",y="+y);return x + y;});System.out.println(sum);// 获取 最大值Integer max = Stream.of(4, 5, 3, 9).reduce(0, (x, y) -> {return x > y ? x : y;});System.out.println(max);// concatStream.concat(Stream.of("1","2"),Stream.of("3","4")).collect(Collectors.toList()).forEach(System.out::println);}

2.3 map

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

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

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

    @Testpublic void singleObj() {// map 操作 // string 转为 integerList<Integer> collect = Stream.of("1", "2", "3").map(Integer::parseInt).collect(Collectors.toList());System.out.println("===========================================");// 修改了person对象的信息List<Person> newPersonList = personList.stream().map(x -> {x.setAge(99);return x;}).collect(Collectors.toList());newPersonList.forEach(System.out::println);}

2.4 Stream结果收集

    @Testpublic void collectTest() {Stream<String> stringStream = personList.stream().map(Person::getLicenseNumber);
//        List<String> collect = stringStream.collect(Collectors.toList());
//        Set<String> collect1 = stringStream.collect(Collectors.toSet());
//        ArrayList<String> collect2 = stringStream.collect(Collectors.toCollection(ArrayList::new));
//        HashSet<String> collect3 = stringStream.collect(Collectors.toCollection(HashSet::new));
//        Object[] objects = stringStream.toArray();for (String s : stringStream.toArray(new IntFunction<String[]>() {@Overridepublic String[] apply(int value) {System.out.println(value);return new String[value];}})) {System.out.println(s);}}
}

2.5 对流中的数据做聚合计算

    @Testpublic void aggregation() {// 最大值System.out.println(personList.stream().collect(Collectors.maxBy((p1, p2) -> {return p1.getAge() - p2.getAge();})).get());// 最小值System.out.println(personList.stream().collect(Collectors.minBy((p1, p2) -> {return p1.getAge() - p2.getAge();})).get());// 平均值System.out.println(personList.stream().collect(Collectors.averagingInt(Person::getAge)));// 数量System.out.println(personList.stream().collect(Collectors.counting()));}

2.6 对流中数据做分组操作

    @Testpublic void listToMap() {// List<Person> 转 Map<String, Person>Map<String, Person> map = personList.stream().collect(Collectors.toMap(Person::getLicenseNumber, Function.identity(), (k1, k2) -> k2));map.forEach((k, val) -> {System.out.println(k + "=" + val);});Map<String, String> personMap = personList.stream().collect(Collectors.toMap(Person::getName, Person::getGender));System.out.println("===========================================");// List<Person> 转 Map<String, List<Person>>Map<String, List<Person>> map2 = personList.stream().collect(Collectors.groupingBy(Person::getGender));map2.forEach((k, val) -> {System.out.println(k + "=" + val);});System.out.println("===========================================");// List<Person> 转 Map<String, Long>Map<String, Long> map3 = personList.stream().collect(Collectors.groupingBy(Person::getGender, Collectors.counting()));map3.forEach((k, val) -> {System.out.println(k + "=" + val);});}

结果：


110101199003070767=Person(name=张宁8, age=18, gender=女, profession=python工程师, licenseNumber=110101199003070767)
110101199603072838=Person(name=张三2, age=32, gender=男, profession=java工程师, licenseNumber=110101199603072838)
110101199003073087=Person(name=张宁6, age=21, gender=女, profession=品质工程师, licenseNumber=110101199003073087)
110101199003071946=Person(name=张宁9, age=27, gender=女, profession=前端工程师, licenseNumber=110101199003071946)
110101199003073407=Person(name=张宁5, age=23, gender=女, profession=品质工程师, licenseNumber=110101199003073407)
===========================================
女=[Person(name=张宁5, age=23, gender=女, profession=品质工程师, licenseNumber=110101199003073407), Person(name=张宁6, age=21, gender=女, profession=品质工程师, licenseNumber=110101199003073087), Person(name=张宁7, age=30, gender=女, profession=java工程师, licenseNumber=110101199003073861), Person(name=张宁8, age=18, gender=女, profession=python工程师, licenseNumber=110101199003070767), Person(name=张宁9, age=27, gender=女, profession=前端工程师, licenseNumber=110101199003071946)]
男=[Person(name=张三1, age=23, gender=男, profession=品质工程师, licenseNumber=110101199603074454), Person(name=张三1, age=26, gender=男, profession=品质工程师, licenseNumber=110101199603074454), Person(name=张三2, age=32, gender=男, profession=java工程师, licenseNumber=110101199603072838), Person(name=张三3, age=14, gender=男, profession=python工程师, licenseNumber=110101199603074315), Person(name=张三4, age=23, gender=男, profession=前端工程师, licenseNumber=110101199603078375)]
===========================================
女=5
男=5

多级分组

        // 多级分组Map<String, Map<Integer, List<Person>>> collect = personList.stream().collect(Collectors.groupingBy(Person::getGender, Collectors.groupingBy(person -> {boolean b = person.getAge() > 25;if (b) {return 88;} else {return 0;}})));collect.forEach((k,v)->{if (CollUtil.isNotEmpty(v)){v.forEach((k1,v1)->{System.out.println(k1+"="+v1);});}});

结果

0=[Person(name=张宁5, age=23, gender=女, profession=品质工程师, licenseNumber=110101199003073407), Person(name=张宁6, age=21, gender=女, profession=品质工程师, licenseNumber=110101199003073087), Person(name=张宁8, age=18, gender=女, profession=python工程师, licenseNumber=110101199003070767)]
88=[Person(name=张宁7, age=30, gender=女, profession=java工程师, licenseNumber=110101199003073861), Person(name=张宁9, age=27, gender=女, profession=前端工程师, licenseNumber=110101199003071946)]
0=[Person(name=张三1, age=23, gender=男, profession=品质工程师, licenseNumber=110101199603074454), Person(name=张三3, age=14, gender=男, profession=python工程师, licenseNumber=110101199603074315), Person(name=张三4, age=23, gender=男, profession=前端工程师, licenseNumber=110101199603078375)]
88=[Person(name=张三1, age=26, gender=男, profession=品质工程师, licenseNumber=110101199603074454), Person(name=张三2, age=32, gender=男, profession=java工程师, licenseNumber=110101199603072838)]

2.7 对流中的数据做分区操作

Collectors.partitioningBy会根据值是否为true,把集合中的数据分割为两个列表，一个true列表，一个false列表

    /*** 对流中数据做分组操作*/@Testpublic void partitioningBy() {Map<Boolean, List<Person>> listMap = personList.stream().collect(Collectors.partitioningBy(x -> x.getAge() > 30));listMap.forEach((k, v) -> {System.out.println(k + "=" + v);});}

结果

false=[Person(name=张三1, age=23, gender=男, profession=品质工程师, licenseNumber=110101199603074454), Person(name=张三1, age=26, gender=男, profession=品质工程师, licenseNumber=110101199603074454), Person(name=张三3, age=14, gender=男, profession=python工程师, licenseNumber=110101199603074315), Person(name=张三4, age=23, gender=男, profession=前端工程师, licenseNumber=110101199603078375), Person(name=张宁5, age=23, gender=女, profession=品质工程师, licenseNumber=110101199003073407), Person(name=张宁6, age=21, gender=女, profession=品质工程师, licenseNumber=110101199003073087), Person(name=张宁7, age=30, gender=女, profession=java工程师, licenseNumber=110101199003073861), Person(name=张宁8, age=18, gender=女, profession=python工程师, licenseNumber=110101199003070767), Person(name=张宁9, age=27, gender=女, profession=前端工程师, licenseNumber=110101199003071946)]
true=[Person(name=张三2, age=32, gender=男, profession=java工程师, licenseNumber=110101199603072838)]

2.8 对流中的数据做拼接

Collectors.joining会根据指定的连接符，将所有的元素连接成一个字符串

    /*** 对流中的数据做拼接*/@Testpublic void join() {System.out.println(personList.stream().map(Person::getLicenseNumber).collect(Collectors.joining(";")));}

2.9 并行的Stream流

前面使用的stream流是串行的，也就是在一个线程上面执行。现在介绍一下并行的stream流，在多个线程中执行

    /*** 并行流使用*/@Testpublic void parallelStream(){// 串行Stream.of(2,8,1,9,9,10).filter(x->{System.out.println(Thread.currentThread().getName()+x);return x>2;}).count();System.out.println("=======================================");// 并行Stream.of(2,8,1,9,9,10)// 将流转换为并发流，Stream处理的时候就会通过多线程处理.parallel().filter(x->{System.out.println(Thread.currentThread().getName()+x);return x>2;}).count();}

结果

main2
main8
main1
main9
main9
main10
=======================================
main9
ForkJoinPool.commonPool-worker-29
main10
main1
ForkJoinPool.commonPool-worker-18
ForkJoinPool.commonPool-worker-22

2.9.1 线程安全问题

    @Testpublic void test01(){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());}

结果

1000java.lang.ArrayIndexOutOfBoundsExceptionat sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)at java.lang.reflect.Constructor.newInstance(Constructor.java:423)at java.util.concurrent.ForkJoinTask.getThrowableException(ForkJoinTask.java:598)at java.util.concurrent.ForkJoinTask.reportException(ForkJoinTask.java:677)at java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:735)at java.util.stream.ForEachOps$ForEachOp.evaluateParallel(ForEachOps.java:160)at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateParallel(ForEachOps.java:174)at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233)at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418)at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:583)at com.mashibing.stream.StreamTest.test01(StreamTest.java:208)at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)at java.lang.reflect.Method.invoke(Method.java:498)at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)at org.junit.runners.ParentRunner.run(ParentRunner.java:363)at org.junit.runners.Suite.runChild(Suite.java:128)at org.junit.runners.Suite.runChild(Suite.java:27)at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)at org.junit.runners.ParentRunner.run(ParentRunner.java:363)at org.junit.runner.JUnitCore.run(JUnitCore.java:137)at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)at com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:33)at com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:230)at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:58)
Caused by: java.lang.ArrayIndexOutOfBoundsException: 366at java.util.ArrayList.add(ArrayList.java:459)at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1374)at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)at java.util.stream.ForEachOps$ForEachTask.compute(ForEachOps.java:291)at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:731)at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)at java.util.concurrent.ForkJoinPool$WorkQueue.execLocalTasks(ForkJoinPool.java:1040)at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1058)at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)

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

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

   @Testpublic void test02() {List<Integer> list = new ArrayList<>();for (int i = 0; i < 1000; i++) {list.add(i);}// 加锁Object o = new Object();List<Integer> listNew = new ArrayList<>();// 使用并行流来向集合中添加数据list.parallelStream().forEach(s -> {synchronized (o) {listNew.add(s);}});System.out.println(listNew.size());// 使用支持并发的集合ConcurrentSkipListSet<Integer> integers = new ConcurrentSkipListSet<>();list.parallelStream().forEach(s -> {integers.add(s);});System.out.println(integers.size());// 保证集合为安全的List<Integer> objects = Collections.synchronizedList(new ArrayList<>());list.parallelStream().forEach(s -> {objects.add(s);});System.out.println(objects.size());// collect操作List<Integer> collect = list.parallelStream().collect(Collectors.toList());System.out.println(collect.size());}

参考文章：https://mp.weixin.qq.com/s/ewUV2-HG2O8xt9dGLUuraA