Java Stream 解析和使用技巧

Stream 的类型
Stream 有分普通流和数值流,之间没有继承关系,普通流用一个泛型表示流中的数据结构类型,如 Stream
数值流主要是避免重复的装箱拆箱,统一用原始数值类型(无法应用泛型指定类型),int long double,我们在做终结操作的时候需要统一装箱 .box() 转成普通流

Stream 的生命周期
创建流 -> 中间操作 -> 终结操作

Stream 的特点
无存储。stream不是一种数据结构,它只是某种数据源的一个视图,数据源可以是一个数组,Java 容器或 I/O channel 等。
为函数式编程而生。对stream的任何修改都不会修改背后的数据源,比如对stream执行过滤操作并不会删除被过滤的元素,而是会产生一个不包含被过滤元素的新stream。
惰式执行。stream上的操作(中间操作)并不会立即执行,只有等到用户真正需要结果的时候(终结操作)才会执行。
可消费性。stream只能被“消费”一次,一旦遍历过就会失效(终结操作就是消费操作),就像容器的迭代器那样,想要再次遍历必须重新生成。
区分中间操作和结束操作最简单的方法,就是看方法的返回值,返回值为stream的大都是中间操作,否则是结束操作。

创建流
从 Colletion
.stream()
.parallelStream()
从数组
Arrays.stream(T array)
Stream.of()
从输入流
BufferedReader.lines()
从目录树
Files.walk(Paths.get(“C:\“))
创建各种数值流
Random.ints()
IntStream.of()
IntStream.range()
…Stream.*()
自己创建流(可创建无穷流)
Stream.generate() 丢进一个类似迭代器的东西即可
Stream.iterate(0, n -> n + 3).limit(10). forEach(x -> System.out.print(x + “ “)); 创建一个自己迭代的流
中间操作
并行化
.parallel()
装箱操作
.boxed() 把数值流转回普通流,才能执行终结操作
转换操作
一对一普通转换 .map()
一对多转换 .flatMap() 本质上是把每个对象转换成流,流会自动合并
Stream<List> stream = Stream.of(Arrays.asList(1,2), Arrays.asList(3, 4, 5));
stream.flatMap(list -> list.stream())

.forEach(i -> System.out.println(i));

直接转成 数值流 .mapToInt .flatMapToInt
4) 排序操作 .sorted

5) 对每一个对象操作 .peek

6) 保留前 n 项 .limit()
无穷流必须执行限流操作,否则将进入死循环

7) 去掉前 n 项 .skip()

8) 筛选操作 .filter()
true 留,false 被删除

终结操作
终结操作后 Stream 将会被消费完成,不能再执行中间操作

转数组 .toArray()
stream.toArray(String[]::new)
转 Collection/String .collect()
forEach 逐一消费所有项目
无法提前结束循环,只能用 return 提前结束当前循环
两两结合操作 .reduce()
.max
.min
.findFirst
.findAny
match 检查
allMatch:Stream 中全部元素符合传入的 predicate,返回 true
anyMatch:Stream 中只要有一个元素符合传入的 predicate,返回 true
noneMatch:Stream 中没有一个元素符合传入的 predicate,返回 true
reduce 操作
reduce操作可以实现从一组元素中生成一个值,sum()、max()、min()、count()等都是reduce操作,将他们单独设为函数只是因为常用。reduce()的方法定义有三种重写形式:

Optional reduce(BinaryOperator accumulator)
T reduce(T identity, BinaryOperator accumulator)
U reduce(U identity, BiFunction<U,? super T,U> accumulator, BinaryOperator combiner)
虽然函数定义越来越长,但语义不曾改变,多的参数只是为了指明初始值(参数identity),或者是指定并行执行时多个部分结果的合并方式(参数combiner)。reduce()最常用的场景就是从一堆值中生成一个值。用这么复杂的函数去求一个最大或最小值,你是不是觉得设计者有病。其实不然,因为“大”和“小”或者“求和”有时会有不同的语义。而Optional是(一个)值的容器,可以避免 null 值的问题,下面会提到。

需求:从一组单词中找出最长的单词。这里“大”的含义就是“长”。

// 找出最长的单词
Stream stream = Stream.of(“I”, “love”, “you”, “too”);
Optional longest = stream.reduce((s1, s2) -> s1.length()>=s2.length() ? s1 : s2);
//Optional longest = stream.max((s1, s2) -> s1.length()-s2.length());
System.out.println(longest.get());
需求:求出一组单词的长度之和。这是个“求和”操作,操作对象输入类型是String,而结果类型是Integer。

// 求单词长度之和
Stream stream = Stream.of(“I”, “love”, “you”, “too”);
Integer lengthSum = stream.reduce(0, // 初始值 // (1)

    (sum, str) -> sum+str.length(), // 累加器 // (2)
    (a, b) -> a+b); // 部分和拼接器,并行执行时才会用到 // (3)

// int lengthSum = stream.mapToInt(str -> str.length()).sum();
System.out.println(lengthSum);
Collect 操作
Collect 是终结操作的一个函数,最为强大,不仅可以将流转化成各种数据结构,也可以再补充中间操作不能进行许多操作。

收集器(Collector)是为Stream.collect()方法量身打造的工具接口(类)。考虑一下将一个Stream转换成一个容器(或者Map)需要做哪些工作?我们至少需要两样东西:

目标容器是什么?是ArrayList还是HashSet,或者是个TreeMap。
新元素如何添加到容器中?是List.add()还是Map.put()。
如果并行的进行规约,还需要告诉collect() 3. 多个部分结果如何合并成一个。

结合以上分析,collect()方法定义为 R collect(Supplier supplier, BiConsumer<R,? super T> accumulator, BiConsumer<R,R> combiner),三个参数依次对应上述三条分析。不过每次调用_collect()都要传入这三个参数太麻烦,收集器Collector 就是对这三个参数的简单封装,所以_collect()的另一定义为<R,A> R collect(Collector<? super T,A,R> collector)。Collectors工具类可通过静态方法生成各种常用的 Collector。

举例来说,如果要将Stream规约成List可以通过如下两种方式实现:

List list = stream.collect(ArrayList::new, ArrayList::add, ArrayList::addAll);// 方式1
常用的转 Collection / String,Collectors 为辅助类

转 list stream.collect(Collectors.toList());
转 set stream.collect(Collectors.toSet());
转其他 stream.collect(Collectors.toCollection(Stack::new));
转 String stream.collect(Collectors.joining()).toString();
转 map

前面已经说过Stream背后依赖于某种数据源,数据源可以是数组、容器等,但不能是Map。反过来从Stream生成Map是可以的,但我们要想清楚Map的key和value分别代表什么,根本原因是我们要想清楚要干什么。通常在三种情况下collect()的结果会是Map:

使用Collectors.toMap()生成的收集器,用户需要指定如何生成Map的key和value。
使用Collectors.partitioningBy()生成的收集器,对元素进行二分区操作时用到。
使用Collectors.groupingBy()生成的收集器,对元素做group操作时用到。
情况 1:使用toMap()生成的收集器,这种情况是最直接的,前面例子中已提到,这是和Collectors.toCollection()并列的方法。如下代码展示将学生列表转换成由<学生,GPA>组成的Map。。

// 使用toMap()统计学生GPA
Map<Student, Double> studentToGPA =

 students.stream().collect(Collectors.toMap(Functions.identity(),// 如何生成key
                                 student -> computeGPA(student)));// 如何生成value

// Functions.identity() 是一个接口默认方法,return x->x,即它本身,在这里是 student -> student
情况 2:使用partitioningBy()生成的收集器,这种情况适用于将Stream中的元素依据某个二值逻辑(满足条件,或不满足)分成互补相交的两部分,比如男女性别、成绩及格与否等。下列代码展示将学生分成成绩及格或不及格的两部分。拉出来之后用 get(true) 和 get(false) 拉出去两个列表。

// Partition students into passing and failing
Map<Boolean, List> passingFailing = students.stream()

     .collect(Collectors.partitioningBy(s -> s.getGrade() >= PASS_THRESHOLD));

情况 3:使用groupingBy()生成的收集器,这是比较灵活的一种情况。跟 SQL 中的group by语句类似,这里的groupingBy()也是按照某个属性对数据进行分组,属性相同的元素会被对应到_Map 的同一个_key上。下列代码展示将员工按照部门进行分组:

// Group employees by department
Map<Department, List> byDept = employees.stream()

        .collect(Collectors.groupingBy(Employee::getDepartment));

以上只是分组的最基本用法,有些时候仅仅分组是不够的。在 SQL 中使用group by是为了协助其他查询,比如1. 先将员工按照部门分组,2. 然后统计每个部门员工的人数。Java 类库设计者也考虑到了这种情况,增强版的groupingBy()能够满足这种需求。增强版的groupingBy()允许我们对元素分组之后再执行某种运算,比如求和、计数、平均值、类型转换等。这种先将元素分组的收集器叫做上游收集器,之后执行其他运算的收集器叫做下游收集器(downstream Collector)。我们可以简单理解,下游收集器就是对 map 的 values 做了一个 forEach

// 使用下游收集器统计每个部门的人数
Map<Department, Integer> totalByDept = employees.stream()

                .collect(Collectors.groupingBy(Employee::getDepartment,
                // 变成 Map<Department, List<Employee>>
                // 对每一个 List<Employee> 执行
                                               Collectors.counting()));// 下游收集器

上面代码的逻辑是不是越看越像 SQL?高度非结构化。还有更狠的,下游收集器还可以包含更下游的收集器,这绝不是为了炫技而增加的把戏,而是实际场景需要。考虑将员工按照部门分组的场景,如果我们想得到每个员工的名字(字符串),而不是一个个_Employee对象_,可通过如下方式做到:

// 按照部门对员工分布组,并只保留员工的名字
Map<Department, List> byDept = employees.stream()

            .collect(Collectors.groupingBy(Employee::getDepartment,
                    // Map<Department, Stream<Employee>>
                    // 对每一个 Stream<Employee> ,执行 mapping,会遍历流中每一个数据
                    Collectors.mapping(Employee::getName,// 下游收集器
                    // 得到一个 Map<Department, Stream<String>>
                            Collectors.toList())));// 更下游的收集器
                    // 得到 Map<Department, List<String>>
                    // Map的key不用管,自动只处理 value 的 stream

Optional 容器
一般用法:
新建一个 可空 Optional,ifPresent 非空则执行 xxx 操作
Optional.ofNullable(text).ifPresent(System.out::println);

从 reduce 等 stream 终结函数返回
2) 检查 Optional 是否为空,一般和三元符配合使用,可同时照顾到非空和空

isPresent()?1:0;

orElse() 用法,取值,如果为空,则为默认值(默认值马上获得,传入的是真实值)
String name = Optional.ofNullable(nullName).orElse(“john”);

orElseGet() 取值,如果为空,则为默认值,默认值为一个获取方法
Optional.ofNullable(text).orElseGet(this::getMyDefault);
当容器内的值为 null 时,orElse() 和 orElseGet() 完全相同,当容器内值不为 null 时,则 orElseGet() 不会执行相关的函数

.filter() 过滤,如果.filter()内容为真,则返回内容,如果为假,则容器内为空。支持链式操作
boolean is2017 = yearOptional.filter(y -> y == 2017).isPresent();
.map() 转换,不用判断非空
int size = listOptional .map(List::size).orElse(0);
.flatMap() 多层 Optional 自动拆开
Stream 底层实现
Stream 实际上是一个流水线(Pipelines),那么他的链式调用+惰性执行的原理是什么呢?

所谓流水线,就是先装配,后启动,一次完成。而不是一步一步迭代实现,这样最大的弊端是没有办法应对复杂的数据结构。效率也十分低

我们举个例子

List test = Arrays.asList(“liu”,”zhang”,”huang”,”chen”,”lix”,”fuc”);

Stream t = test.stream();
Stream t2 = t.skip(2);
Stream t3 = t2.map(x -> x.substring(2));
Stream t4 = t3.sorted();
String t5 = t4.max(String::compareTo).orElse(“”);
这是一组流水线 Stream 拆开来生成多个 Stream 变量。我们知道,Stream 实际上是一个接口,那么,我们调用了这些函数之后,到底返回了一个什么对象呢?我们直接用 IDE 告诉我们答案

可以看到,首先这里有一个双向链表的结构,每次中间操作,都会增加一个新的 AbstractPipeline,然后记录第一个 AbstractPipeline 和 上一个 AbstractPipeline,上一个 AbstractPipeline 也会记录当前新增的 AbstractPipeline。

而另一方面,根据增加的操作不同,也会有不同的 AbstractPipeline 子类,包括 ReferencePipeline, SliceOps, SortedOps, StatelessOp 等等,只是实现的层级不同,我们稍后在纠结这些。

并且,这些实现类内部会有一个 核心的逻辑方法opWrapSink(int flags, Sink<P_OUT> sink,会把逻辑打包成一个 Sink 对象,这个 Sink 对象还接收另外一个 Sink 对象作为构造函数参数。

我们拿 .filter() 举例,内置了一个函数会返回 Sink 对象,目前还是惰性执行,所以没有立刻生成:

@Override

public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
    Objects.requireNonNull(predicate);
    return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
                                 StreamOpFlag.NOT_SIZED) {
        @Override
        Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
            return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {
                @Override
                public void begin(long size) {
                    downstream.begin(-1);
                }

                @Override
                public void accept(P_OUT u) {
                    if (predicate.test(u))
                        downstream.accept(u);
                }
            };
        }
    };
}

Sink 对象源码,我们最关注的是构造函数,可以看到它又藏了另外一个 sink

static abstract class ChainedReference<T, E_OUT> implements Sink {

    protected final Sink<? super E_OUT> downstream;

    public ChainedReference(Sink<? super E_OUT> downstream) {
        this.downstream = Objects.requireNonNull(downstream);
    }

    @Override
    public void begin(long size) {
        downstream.begin(size);
    }

    @Override
    public void end() {
        downstream.end();
    }

    @Override
    public boolean cancellationRequested() {
        return downstream.cancellationRequested();
    }
}

当我们走到终结操作的时候,会先执行一个这样的操作:

final <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) {

Objects.requireNonNull(sink);
// 检查非空
for ( @SuppressWarnings("rawtypes") AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) {
    // 从后向前调用每个AbstractPipeline的opwrapSink,然后每个 Sink 藏着上一个 Sink
    sink = p.opWrapSink(p.previousStage.combinedFlags, sink);
}
return (Sink<P_IN>) sink;

}
好了,千辛万苦,我们终于得到了这么一个 Sink,这个 Sink 保存了所有的中间流操作和最后一个 reduce 规约操作的所有操作对象。也就是说,我们的流水线建成了。

拿到这个 Sink 之后,我们就可以愉快的进行迭代了

// AbstractPipelie.copyInto()
final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {

Objects.requireNonNull(wrappedSink);

if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
    wrappedSink.begin(spliterator.getExactSizeIfKnown());// 调用遍历前钩子,通知数据大小
    spliterator.forEachRemaining(wrappedSink);// 迭代器
    wrappedSink.end();// 调用遍历后钩子
}
else {
    copyIntoWithCancel(wrappedSink, spliterator);
}

}
遍历调用 Sink 的 begin() 钩子,主要是用来准备数据结构,每个 Sink 的 begin 都会递归调用下游的 begin

// SliceOps
public void begin(long size) {

downstream.begin(calcSize(size, skip, m));

}
调用 forEachRemaining() 方法

// Spliterators
if ((a = array).length >= (hi = fence) &&

(i = index) >= 0 && i < (index = hi)) {
do { action.accept((T)a[i]); } while (++i < hi);

}
// 就是这么简单粗暴,把 Stream 里面的东西一个一个调用 Sink 里面的 accept 方法
// 然后,这个东西又会调用下游的 accept
// SliceOps
@Override
public void accept(T t) {

if (n == 0) {
    if (m > 0) {
        m--;
        downstream.accept(t);
    }
}
else {
    n--;
}

}
// 这里,我们可以看到,切割操作就是有的元素不往下传,就gg了,往下传就继续下面的 accept()
最后调用 end() 方法封口,同样是递归调用

// SotedOps()
// Sort有他的特殊性,不能在 accept 的时候一个一个执行,只能在封口的时候,再排序
public void end() {

list.sort(comparator); // 排序
downstream.begin(list.size()); // 通知下游准备
if (!cancellationWasRequested) { // 下游Sink不包含短路操作
    list.forEach(downstream::accept); // 把元素继续一个一个丢给下游
}
else {
    for (T t : list) { //把元素一个一个拉出来
        if (downstream.cancellationRequested()) break;// 每次都调用cancellationRequested()询问是否可以结束处理。
        downstream.accept(t); //否则,把这个交给下游
    }
}
downstream.end(); // 调用下游的 end()函数
list = null;

}
// 来看下它的其他两个操作,可以看到,他并没有调用下游操作,而是直接拦截了,等到 end 的时候再通知下游
@Override
public void begin(long size) {

...
// 创建一个存放排序元素的列表
list = (size >= 0) ? new ArrayList<T>((int) size) : new ArrayList<T>();

}
@Override
public void accept(T t) {

// Sorted的违规操作,
list.add(t);// 1. 使用当前Sink包装动作处理t,只是简单的将元素添加到中间列表当中

}
经过这些处理之后,会被丢进 reduce 操作 或者是 collect 操作收集 流中的数据。

关于并且流时候的情况,调用了 Fork/Join 框架,比较复杂,以后再更。

原创文章:https://www.qqhhs.com,作者:起航®,如若转载,请注明出处:https://www.qqhhs.com/69.html

版权声明:本站提供的一切软件、教程和内容信息仅限用于学习和研究目的,请于下载的24小时内删除;不得将上述内容用于商业或者非法用途,否则,一切后果请用户自负。本站信息来自网络收集整理,如果您喜欢该程序和内容,请支持正版,购买注册,得到更好的正版服务。我们非常重视版权问题,如有侵权请邮件与我们联系处理。敬请谅解!

Like (0)
Donate 受君之赠,荣幸之至 受君之赠,荣幸之至 受君之赠,荣幸之至 受君之赠,荣幸之至
Previous 2023年1月10日
Next 2023年1月10日

相关推荐

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

联系我们

qhhl

QQ-175142992024110802215838同号

SHARE
TOP
“人们不愿意相信,一个土匪的名字叫牧之”