Java stream使用与执行原理

2024-09-08 04:28
文章标签 java 使用 原理 执行 stream

本文主要是介绍Java stream使用与执行原理,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

stream简介

Stream: A sequence of elements supporting sequential and parallel aggregate operations

stream为sequential即单线程串行操作,parallelStream支持并行操作,本文只讨论sequential的stream。

stream常用操作

    @Datastatic class Course {private Long number;private LocalDateTime beginTime;private List<Long> studentIds;}public void streamOperations(List<Course> courses) {// 映射并去重List<Long> courseNumbers = courses.stream().filter(Objects::nonNull).map(Course::getNumber).distinct().collect(Collectors.toList());// 先按开始时间排序后按number排序List<Course> sortedCourses = courses.stream().sorted(Comparator.comparing(Course::getBeginTime).thenComparing(Course::getNumber)).collect(Collectors.toList());// 根据number组成map, 如果有相同的number会抛异常Map<Long, Course> num2Lesson1 = courses.stream().collect(Collectors.toMap(Course::getNumber, Function.identity()));// 根据number组成map, 如果有相同的number会执行降级逻辑Map<Long, Course> num2Lesson2 = courses.stream().collect(Collectors.toMap(Course::getNumber, Function.identity(), (v1, v2) -> v1));// 根据number聚合Map<Long, List<Course>> num2Lessons = courses.stream().filter(Objects::nonNull).collect(Collectors.groupingBy(Course::getNumber));// 根据number聚合某个字段Map<Long, List<LocalDateTime>> number2BeginTimes = courses.stream().filter(Objects::nonNull).collect(Collectors.groupingBy(Course::getNumber,Collectors.mapping(Course::getBeginTime, Collectors.toList())));// 根据number找到number下最大beginTime的CourseMap<Long, Optional<Course>> number2MaxBeginTimeCourse = courses.stream().filter(r -> Objects.nonNull(r.getBeginTime())).collect(Collectors.groupingBy(Course::getNumber, Collectors.maxBy(Comparator.comparing(Course::getBeginTime))));// 获取course下所有的studentIdList<Long> allStudentIds = courses.stream().map(Course::getStudentIds).flatMap(Collection::stream).distinct().collect(Collectors.toList());}

stream原理

基本原理

        list.stream().filter(Objects::nonNull).map(World::toString).distinct().collect(Collectors.toList());

以上面的处理为例,分别经过了过滤->映射->去重->聚合三个操作,在stream内部会通过一个链表将这三个操作联系起来,一个操作被称为一个stage(或pipeline),每个stage会指向上下游的stagesourceStage(即哨兵头节点),如下图所示:

在这里插入图片描述

对应的在AbstractPipeline类中有三个字段分别引用链表上下游节点和链表的哨兵头节点:

abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>extends PipelineHelper<E_OUT> implements BaseStream<E_OUT, S> {// Backlink to the head of the pipeline chain (self if this is the source stage).private final AbstractPipeline sourceStage;// The "upstream" pipeline, or null if this is the source stage.private final AbstractPipeline previousStage;  // The next stage in the pipeline, or null if this is the last stage. Effectively final at the point of linking to the next pipeline.      private AbstractPipeline nextStage;  ...    
}

stage可分为3类(可以在各个Reference类中找到下面3个内部类):

  • Header: 哨兵头节点,用户无需感知
  • StatelessOp: 无状态stage,如过滤
  • StatefulOp: 有状态stage,如聚合

对应的在ReferencePipeline中有3个内部类:

abstract class ReferencePipeline<P_IN, P_OUT>extends AbstractPipeline<P_IN, P_OUT, Stream<P_OUT>>implements Stream<P_OUT>  {static class Head<E_IN, E_OUT> extends ReferencePipeline<E_IN, E_OUT> {...}abstract static class StatelessOp<E_IN, E_OUT> extends ReferencePipeline<E_IN, E_OUT> {...}abstract static class StatefulOp<E_IN, E_OUT> extends ReferencePipeline<E_IN, E_OUT> {...}  ...     
}

以上提到的三种名词:pipeline,stage,op 都是指代链表里的一个操作节点,即 pipeline == stage == op,类似一个生物学人具有多个社会学身份。

再来看看代码实现,其uml类图如下:

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传
Java对BaseStream接口的实现是 AbstractPipelineBaseStream可分为基于引用类型和基于基础类型,其中基于引用类型实现为ReferencePipleline,基于数值基础类型分别有实现LongPiplelineIntPiplelineDoublePipleline

pipeline中封装了stream sourceintermediate operations,一个pipeline代表一次操作,比如过滤、去重等,当 pipeline 被引用时则称为stage,多个stage可以通过Fluent Api组装起来实现流式处理,组装的过程即是构建一个链表的过程。

当我们调用一个集合的stream()方法时,会调用StreamSupport#stream方法构造一个header pipeline:

public interface Collection<E> extends Iterable<E> {...default Stream<E> stream() {// 传入Collection自定义个Spliterator,返回一个 header pipelinereturn StreamSupport.stream(spliterator(), false);}...// Collection自定义的Spliteratordefault Spliterator<E> spliterator() {return Spliterators.spliterator(this, 0);}...
}public final class StreamSupport {...// 生成 header pipelinepublic static <T> Stream<T> stream(Spliterator<T> spliterator, boolean parallel) {Objects.requireNonNull(spliterator);return new ReferencePipeline.Head<>(spliterator,StreamOpFlag.fromCharacteristics(spliterator),parallel);}...
}
java.util.Spliterator

两个问题:

  1. Spliterator是干嘛的?
  2. stream为什么需要Spliterator?

An object for traversing and partitioning elements of a source.

可以看到Spliterator支持对数据进行遍历和分割,对应的在接口中有tryAdvance + forEachRemaining用于遍历,有trySplit支持分割。

trySplit方法返回的是Spliterator,所以Spliterator是一种类似细胞分裂的方式执行,对一个ArrayList进行分割:

        List<Integer> list = new ArrayList<>();for (int i = 1; i <= 5; i++) {list.add(i);}Spliterator<Integer> sourceSpliterator = list.spliterator();Assertions.assertEquals(5, sourceSpliterator.estimateSize());// 执行一次,输出1,剩下2345四个元素可分割和遍历sourceSpliterator.tryAdvance(i -> Assertions.assertEquals(1, i));Assertions.assertEquals(4, sourceSpliterator.estimateSize());Spliterator<Integer> subSpliterator1 = sourceSpliterator.trySplit();// 2 3Assertions.assertEquals(2, sourceSpliterator.estimateSize());// 4 5Assertions.assertEquals(2, subSpliterator1.estimateSize());List<Integer> list2 = new ArrayList<>();list2.add(1);// 只有一个元素时进行split,此时spliterator1==nullSpliterator<Integer> spliterator1 = list2.spliterator().trySplit();Assertions.assertNull(spliterator1);

Spliterator只对未遍历过的元素(未被tryAdvance执行到且未执行forEachRemaining)执行trySplit,如果没有trySplit返回null, 同样stream流只运行执行一次。

同时Spliterator有以下特性,可以包含多个:

  • ORDERED: 遍历和分割保证顺序
  • DISTINCT: 非重复
  • SORTED: 遍历和分割时以一种顺序执行,通过getComparator方法提供自定义比较器
  • SIZED: estimateSize放回返回固定值
  • SUBSIZED: trySplit之后所有的Spliterator同时支持SIZED和SUBSIZED特性
  • IMMUTABLE: 遍历和分割的对象不能有结构变更
  • CONCURRENT: 支持多线程安全遍历和分割

所有特性以bitset的方式记录在一个int类型值中,通过characteristics方法获取。

那么为什么stream要用Spliterator呢?

Spliterator是并行流(Parallel Stream)背后的关键机制。当调用集合的parallelStream()方法时,该方法内部会创建一个Spliterator来遍历和分割集合中的元素。然后,Java的并行框架(如ForkJoinPool)会利用这些Spliterator来分配任务给多个线程,以实现并行处理。

java.util.stream.Sink

stream的操作都在该接口中实现

An extension of Consumer used to conduct values through the stages of a stream pipeline,
with additional methods to manage size information, control flow, etc.

通常使用内部抽象类ChainedReference构建一个Sink链,ChainedReference 中指向链条的下一个Sink
stream支持多元素操作如sorted和单元素操作如map,如何组合这两种操作呢?stream即是通过Sink接口实现。

Sink包含三个主要接口:

interface Sink<T> extends Consumer<T> {// 调用该接口表示stage开始接收数据,size表示要接受的数据个数,-1表示未知或无限制default void begin(long size) {}// 调用该接口表示stage数据接受完毕,当需要操作所有数据时,可在这里操作,比如sorted就在这里做排序default void end() {}// 调用该接口表示stage开始操作单个数据default void accept(int value)...
}

注意以上接口都是default,如果子接口(如TerminalSink)没实现表示默认不做操作。

以以下stream流为例:


list = [3,2,5]list.stream().filter(Objects::nonNull).map(i -> i + "hello").distinct().sorted().forEach(System.out::println);

当我们调用list.stream.filter.map.distinct.sorted.collect时,
会首先正向构建一个stage操作双向链表,即filter <-> map <-> distinct <-> sorted <-> collect
最后在链接TerminalOp类型的stage时(这里是collect)会调用AbstractPipeline#wrapSink方法构建Sink单向链表,Sink单向链表的指向顺序也是filter -> map -> distinct -> sorted -> collect,但其构建顺序是反向的,即collect -> sorted -> distinct -> map -> filter,如下图所示:

在这里插入图片描述

代码如下:

abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>extends PipelineHelper<E_OUT> implements BaseStream<E_OUT, S> {...// .stream()执行时表示中间操作stage的个数// .parallelStream()执行时表示前面有状态的中间操作个数,因为有状态依赖的必须sequential执行private int depth;@Overridefinal <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) {Objects.requireNonNull(sink);for ( @SuppressWarnings("rawtypes") AbstractPipeline p=AbstractPipeline.this; p.depth > 0; // 前面的stagep=p.previousStage) {sink = p.opWrapSink(p.previousStage.combinedFlags, sink);}return (Sink<P_IN>) sink;}...    
}

来模拟[3,2,5]作为输入时的stream流程:
在这里插入图片描述

这篇关于Java stream使用与执行原理的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1147108

相关文章

Linux使用fdisk进行磁盘的相关操作

《Linux使用fdisk进行磁盘的相关操作》fdisk命令是Linux中用于管理磁盘分区的强大文本实用程序,这篇文章主要为大家详细介绍了如何使用fdisk进行磁盘的相关操作,需要的可以了解下... 目录简介基本语法示例用法列出所有分区查看指定磁盘的区分管理指定的磁盘进入交互式模式创建一个新的分区删除一个存

C#使用HttpClient进行Post请求出现超时问题的解决及优化

《C#使用HttpClient进行Post请求出现超时问题的解决及优化》最近我的控制台程序发现有时候总是出现请求超时等问题,通常好几分钟最多只有3-4个请求,在使用apipost发现并发10个5分钟也... 目录优化结论单例HttpClient连接池耗尽和并发并发异步最终优化后优化结论我直接上优化结论吧,

SpringBoot使用Apache Tika检测敏感信息

《SpringBoot使用ApacheTika检测敏感信息》ApacheTika是一个功能强大的内容分析工具,它能够从多种文件格式中提取文本、元数据以及其他结构化信息,下面我们来看看如何使用Ap... 目录Tika 主要特性1. 多格式支持2. 自动文件类型检测3. 文本和元数据提取4. 支持 OCR(光学

Java内存泄漏问题的排查、优化与最佳实践

《Java内存泄漏问题的排查、优化与最佳实践》在Java开发中,内存泄漏是一个常见且令人头疼的问题,内存泄漏指的是程序在运行过程中,已经不再使用的对象没有被及时释放,从而导致内存占用不断增加,最终... 目录引言1. 什么是内存泄漏?常见的内存泄漏情况2. 如何排查 Java 中的内存泄漏?2.1 使用 J

JAVA系统中Spring Boot应用程序的配置文件application.yml使用详解

《JAVA系统中SpringBoot应用程序的配置文件application.yml使用详解》:本文主要介绍JAVA系统中SpringBoot应用程序的配置文件application.yml的... 目录文件路径文件内容解释1. Server 配置2. Spring 配置3. Logging 配置4. Ma

Linux使用dd命令来复制和转换数据的操作方法

《Linux使用dd命令来复制和转换数据的操作方法》Linux中的dd命令是一个功能强大的数据复制和转换实用程序,它以较低级别运行,通常用于创建可启动的USB驱动器、克隆磁盘和生成随机数据等任务,本文... 目录简介功能和能力语法常用选项示例用法基础用法创建可启动www.chinasem.cn的 USB 驱动

Java 字符数组转字符串的常用方法

《Java字符数组转字符串的常用方法》文章总结了在Java中将字符数组转换为字符串的几种常用方法,包括使用String构造函数、String.valueOf()方法、StringBuilder以及A... 目录1. 使用String构造函数1.1 基本转换方法1.2 注意事项2. 使用String.valu

C#使用yield关键字实现提升迭代性能与效率

《C#使用yield关键字实现提升迭代性能与效率》yield关键字在C#中简化了数据迭代的方式,实现了按需生成数据,自动维护迭代状态,本文主要来聊聊如何使用yield关键字实现提升迭代性能与效率,感兴... 目录前言传统迭代和yield迭代方式对比yield延迟加载按需获取数据yield break显式示迭

使用SQL语言查询多个Excel表格的操作方法

《使用SQL语言查询多个Excel表格的操作方法》本文介绍了如何使用SQL语言查询多个Excel表格,通过将所有Excel表格放入一个.xlsx文件中,并使用pandas和pandasql库进行读取和... 目录如何用SQL语言查询多个Excel表格如何使用sql查询excel内容1. 简介2. 实现思路3

java脚本使用不同版本jdk的说明介绍

《java脚本使用不同版本jdk的说明介绍》本文介绍了在Java中执行JavaScript脚本的几种方式,包括使用ScriptEngine、Nashorn和GraalVM,ScriptEngine适用... 目录Java脚本使用不同版本jdk的说明1.使用ScriptEngine执行javascript2.