java Streams API介绍

Posted 2020-11-03 yuanbing1226

java8中引入了java.util.stream这样一个包（流），新特性的添加旨在能帮助开发人员更高的抽象层次上对集合进行一系列操作。
借助java.util.stream包，我们可以简明的声明性的表达集合，数组和其他数据源上可能的并行处理。实现从外部迭代到内部迭代的改变。

更高级别的抽象

考虑这样一个问题，当我们要收集一个班级来自陕西的学生的时候，在java8以前，我们一般都是这样实现的。

        public List<Student> isFromShannxi(List<Student> students){
            Objects.requireNonNull(students, "Cannot find students");
            List<Student> shannxiStudents = new ArrayList<>();
            for(Student student : students) {
                if (student.isFrom(Constants.Province.SHANNXI)) {
                    shannxiStudents.add(student);
                }
            }
            return shannxiStudents;
        }

这样的代码估计每个java开发人员分分钟都可实现。但是这样的操作每次都要写很多样板代码，创建一个保存收集结果的集合，对传进来的集合进行遍历，通过特定的条件筛选，然后将筛选的结果添加到结果集并返回。这样的代码有几个缺点
1 维护困难，如果不能读完整个循环是不能猜出来代码的本意，如果命名不规范或者没有注释（试想一下上边的变量命名都是这样的 xs(学生) lzsxdxs(来自陕西的学生) ...），维护起来更加困难，大家不要以为没有这样的代码，我自己曾经接手过这样的代码，那是相当的痛苦。
2 难以扩展为并行执行，要修改为并行处理估计会话费很大的精力。

下面展示了用流来实现相同的功能

        public List<Student> isFromShannxi(Stream<Student> stream){
            Objects.requireNonNull(stream, "Cannot find students");
            return stream
                    .filter(student -> student.isFrom(Constants.Province.SHANNXI))
                    .collect(toList());
        }

从代码我们可以看到程序的本意了,将stream过滤并将过滤后的结果转为集合并返回，至于转成哪种类型的，由JVM进行推断。整个程序的执行过程像是在叙述一件事，没有过多的中间变量，可以减小GC压力，提高效率。那么接下来开始探索stream的功能。

stream简介

java.util.stream这个包引入了流。
流和集合有以下几个不同的地方。

• 没有存储。流不是存储元素的数据结构，相反，它通过计算操作传递来自诸如数据结构，数组，构造器函数或者I/O通道等源的元素。
  
• 本质上讲，流的操作会产生结果，但不会修改其来源。例如Stream从集合获取一个没有过滤元素的新元素，而不是从集合中删除过滤元素。
  
• 惰性求值，流的许多操作都是惰性的。
  
• 及早求值，在一系列惰性求值之后调用一个及早求值的方法，来生成最终的结果。
  
• 集合有大小限制，但是流没有大小限制。
  
• 流元素在声明周期中只被访问一次，就像迭代器一样，当需要再次访问相同的元素时必须生成新的流。

java中提供了以下方式来生成流

• Collection.stream() 使用集合创建流；
• Collection.parallelStream() 使用集合创建并行流；
• Arrays.stream(Object[]);
• Stream.of(Object[])， IntStream.range(int, int) 或Stream.iterate(Object, UnaryOperator)， Stream.empty(), Stream.generate(Supplier
• BufferedReader.lines();
• Random.ints();
• BitSet.stream()， Pattern.splitAsStream(java.lang.CharSequence)，和JarFile.stream()。

流的中间操作（无副作用）

• filter(Predicate
• map(Function
• flatMap(Function
• distinct()已删除了重复的流元素
• sorted() 按自然顺序排序的流元素
• Sorted(Comparator
• limit(long)截断至所提供长度的流元素
• skip(long)丢弃了前 N 个元素的流元素
• takeWhile(Predicate
• dropWhile(Predicate

终止操作

• forEach(Consumer
• toArray() 使用流的元素创建一个数组。
• reduce(...) 将流的元素聚合为一个汇总值。
• collect(...) 将流的元素聚合到一个汇总结果容器中。
• min(Comparator
• max(Comparator
• count() 返回流的大小。
• {any,all,none}Match(Predicate
• findFirst() 返回流的第一个元素（如果有）。
• findAny()

流操作和管道

流操作作为中间造作和终止操作，并组合起来成为流管道。流管道由一个源（例如 Collection，一个数组，一个生成器函数或一个I / O通道）组成; 接着是零或多个中间操作，例如 Stream.filter或Stream.map; 和诸如Stream.forEach或的终端操作Stream.reduce。诸如Stream.forEach或者的 终止操作IntStream.sum可以遍历流以产生结果或副作用。终止操作执行后，流管道被视为消耗，不能再使用; 如果您需要再次遍历相同的数据源，则必须返回到数据源以获取新的流。
缓慢处理流程可实现显着的效率;可以将过滤，映射和求和融合为数据上的单次通过，并具有最小的中间状态。懒惰还可以避免在没有必要时检查所有数据; 对于诸如“查找长度超过1000个字符的第一个字符串”等操作，只需检查足够的字符串即可找到具有所需特性的字符串，而无需检查源中可用的所有字符串。（当输入流是无限的，而不仅仅是大的时候，这种行为变得更加重要。）中间业务进一步分为无状态 和有状态操作。无状态操作（例如filter 和）map在处理新元素时不会保留先前看到的元素的状态 - 每个元素可以独立于其他元素上的操作进行处理。有状态的操作，比如 distinct和sorted，可能会包含以前看到的元素在处理新元素时的状态。有状态的操作可能需要在生成结果之前处理整个输入。例如，只有在查看了流的所有元素之后，才能对排序流产生任何结果。因此，在并行计算中，一些包含有状态中间操作的管道可能需要对数据进行多次传递，或者可能需要缓存重要数据。只包含无状态中间操作的流水线可以一次处理，无论是顺序处理还是并行处理，只需最少的数据缓冲。此外，有些操作被认为是短路操作。如果在呈现无限输入时，中间操作是短路的，其可能产生有限的流。终端操作是短路的，如果出现无限输入时，它可能在有限的时间内终止。在流水线中进行短路操作是处理无限流在有限时间内正常终止的必要但不充分的条件。

查看一个班级里边来自陕西且选修了电力电子的学生的总数
具有显式for-循环的处理元素本质上是串行的。流通过将计算重新定义为聚合操作的流水线而不是作为每个单独元素的命令操作来促进并行执行。所有流操作都可以以串行或并行方式执行。除非明确要求并行性，否则JDK中的流实现会创建串行流。例如，Collection具有分别产生顺序流和并行流的方法 Collection.stream()和Collection.parallelStream(); 其他流载方法（例如IntStream.range(int, int) 产生顺序流），但是可以通过调用它们的BaseStream.parallel()方法来高效地并行化这些流。要并行执行之前的“小部件权重总和”查询，我们可以这样做

        return students.stream()
                .filter(student -> student.isFrom(Constants.Province.SHANNXI))
                .filter(student -> student.getScores().containsKey(Constants.Course.POWER_ELECTRONICS))
                .count();
            

这个程序执行起来很快，因为每个班级只有几十个学生，但是如果我需要选出全国的在校大学生里边来自陕西的且选修了电力电子课程的学生。那么执行起来就有点久了。这时我们需要将串行改为并行来充分利用多核cpu资源。传统的修改方式需要改成个代码的结构，但是流的修改就显得很简单了。我们只需要添加一行代码就可以了。至于将串行流变成并行流的原理和限制条件，我们后边再分析。

        return students.stream()
                .parallel()
                .filter(student -> student.isFrom(Constants.Province.SHANNXI))
                .filter(student -> student.getScores().containsKey(Constants.Course.POWER_ELECTRONICS))
                .count();
            

这个例子的串行和并行版本之间的唯一区别是创建初始流，使用“ parallelStream()”而不是“ stream()”。当启动终止操作时，根据流调用的流的方向，流管道将按顺序或并行执行。可以使用该isParallel()方法确定流是以串行还是并行方式执行，并且可以使用BaseStream.sequential()和 BaseStream.parallel()操作来修改流的方向 。当启动终止操作时，根据流调用的流的模式顺序或并行执行流管道。除了确定为明确不确定的操作外，例如findAny()流是顺序执行还是并行执行都不应改变计算结果。大多数流操作接受描述用户指定行为的参数，这些参数通常是lambda表达式。为了保持正确的行为，这些行为参数必须是无干扰的，并且在大多数情况下必须是无状态的。这样的参数总是一个实例功能接口如Function，也常常lambda表达式或方法的引用。

不修改数据源

流使你能够在各种数据源上执行可能并行的聚合操作，甚至包括非线程安全的集合，如ArrayList。 这只有在我们可以防止在执行流管线期间干扰数据源时才有可能。 除了escape-hatch操作iterator（）和spliterator（）之外，当调用终端操作时开始执行，并在终端操作完成时结束。 对于大多数数据源来说，防止干扰意味着确保数据源在执行流管道期间根本不被修改。 值得注意的例外是流的源是并发集合，这些集合专为处理并发修改而设计。 并发流源是那些Spliterator报告CONCURRENT特性的流源。如果行为参数的修改导致数据源的修改，则说行为参数会干扰并发数据源。对所有管道流的要求不限于并行管道，除非流源是并发的，否则在执行流管道期间修改流的数据源可能会导致异常，或者结果不一致。对于表现良好的流源，在终止操作之前可以对源进行修改，这些修改可以覆盖之前的流。如下所示。

        public static void modifyStream() {
            List<String> list = new ArrayList<>(Arrays.asList("one", "two"));
            Stream<String> stream = list.stream();
            list.add("three");
            System.out.println(stream.collect(Collectors.joining(", ", "[", "]")));
        }

首先创建一个由两个字符串组成的列表：“one”; 和“two”。然后从该列表创建一个流。接下来通过添加第三个字符串修改列表：“three”。最后，流的元素被收集并连接在一起。由于在终端collect 操作开始之前列表已被修改，结果将是一串"[one, two, three]"。从JDK集合返回的所有流以及大多数其他JDK类以这种方式运行良好;

无状态行为

如果流操作的行为参数是有状态的，则并行化操作的结果可能不一致。有状态的lambda的例子是map（）操作。如果操作是并行的，由于线程调度的不同，相同的输入可能产生不同的结果。而无状态的lambda表达式的结果总是相同的。所以最好的办法是在整个流操作中避免有状态的行为参数。

副作用

行为参数对流操作的副作用通常是不鼓励的，因为它们通常会导致无意中违反无状态要求以及其他线程安全危害。
如果行为参数确实有副作用，除非明确说明，否则不能保证 这些副作用对其他线程的 可见性，也不能保证在同一个流管线内的“相同”元素上进行不同的操作在同一个线程中执行。此外，这些影响的排序可能令人惊讶。即使流水线被限制产生一个与流源碰到次序一致的 结果（例如，IntStream.range(0,5).parallel().map(x -> x*2).toArray() 必须产生[0, 2, 4, 6, 8]），也不能保证映射器函数应用于单个元素的顺序，或者对于给定的元素，什么线程执行任何行为参数。许多计算可能会被诱惑使用副作用，可以更安全有效地表达，而不会产生副作用，例如使用 简化而不是可变累加器。但是，诸如println()用于调试目的的副作用通常是无害的。少量流操作（如 forEach()和）peek()只能通过副作用进行操作; 这些应该小心使用。作为如何将不恰当使用副作用的流管道转换为不适用的流管道的示例，下面的代码搜索匹配给定正则表达式的字符串流，并将匹配放入列表中。

         ArrayList<String> results = new ArrayList<>();
         stream.filter(s -> pattern.matcher(s).matches())
               .forEach(s -> results.add(s));  // Unnecessary use of side-effects!

此代码不必要地使用副作用。如果并行执行，非线程安全性ArrayList会导致不正确的结果，并且添加所需的同步会导致争用，从而破坏并行性的好处。此外，在这里使用副作用是完全不必要的; 在forEach()可以简单地用一个缩小操作是更安全，更有效，并且更适合于并行替换：

         List<String>results =
             stream.filter(s -> pattern.matcher(s).matches())
                   .collect(Collectors.toList());  // No side-effects!

顺序

流可能有也可能没有顺序，流是否有顺序取决于流源和中间操作，某些流源（例如List或数组）是内在排序的，而其他（如HashSet）则不是。一些中间业务，比如sorted()，可以在一个无序的，否则是流的初始顺序，以及其他可能导致一个有序的流无序的，如BaseStream.unordered()。此外，一些终止操作可能会忽略流次序，例如 forEach()。如果流是有序的，大多数操作都被限制为按照它们的遇到顺序操作元素; 如果流的源是一个List contains [1, 2, 3]，那么执行的结果map(x -> x*2) 必须是[2, 4, 6]。但是，如果源没有定义的遇到顺序，值[2, 4, 6]的任何排列都是有效的结果。对于并行流，放宽排序约束有时可以实现更高效的执行。如果元素的顺序不相关，某些集合操作（??如过滤重复项（distinct()）或分组还原（Collectors.groupingBy()））可以更高效地实现。类似地，与碰到命令有内在联系的操作limit()可能需要缓冲来确保正确的排序，从而破坏并行性的好处。在流有碰到命令的情况下，但用户并不特别关心该碰到命令，明确地对该流进行取消unordered()可以提高某些有状态或终端操作的并行性能。然而，即使在排序约束下，大多数流管线仍然有效地并行化。

reduce操作

retuce操作取输入元素的序列，并通过组合操作的反复应用，例如找到一组数字，或累积元素的总和或最大到一个列表它们组合成一个单一的汇总结果。该流的类具有普遍减少操作，所谓的多种形式 reduce() 和collect()，以及多个专业化还原的形式，如 sum()，max()或count()。比如要实现1到100相加。传统for循环是这样实现的。

        int sum = 0;
        for(int x : numbers) {
            sum += x;
        }
        return sum;
        
        return numbers.stream()
                .reduce(0, Integer::sum);
            

可变的reduce

我们想把一个Collection中的字符串拼接起来，可以通过流处理元素。如下

        String result = strings.stream()
                .reduce("", String::concat);

        
        StringBuilder result = strings.stream()
                .collect(StringBuilder::new,
                        (sb, s) -> sb.append(s),
                        (sb, sb2) -> sb.append(sb2));
                        
        StringBuilder result = strings.stream()
                .collect(StringBuilder::new,
                        StringBuilder::append,
                        StringBuilder::append);
                    

上边第一个示例中，虽然可以将string拼接起来，但是由于要不断的new String对象，效率不高。我们应该用StringBuilder来实现这个功能，第二个和第三个是用StringBuilder实现的字符串拼接，其中第三段代码用方法引用对第二段进行了优化。但是这样的感觉还是有点繁琐。能不能有更简单的方式来实现这个功能呢，别担心，类库的设计者已经想到了这种应用场景。Collectors.join()就是专门来解决这个问题的。我们来看用现成API的示例

        String result = strings.stream()
                    .collect(Collectors.joining(""));
            

其中joining方法有几个重载方法，可以添加分割符，前缀和后缀，比如Collectors.joining(", ", "[", "]" ),加入我们测试的字符串是"Hello" "World",则其结果为"[Hello, World]"

结束语

stream提供的功能非常强大，在java.util.stream包下还有个类Collectors，它和stream是好搭档，通过组合来以更简单的方式来实现更加强大的功能。上述代码清单可以在Github下载。

参考文档

java Streams
应该返回流还是集合
java8获取流中的最小值和最大值
java.util.stream
java8 函数式编程