系统学习Stream

2024-03-08 17:14:22

Java8中最大的两个亮点，一个是Lambda表达式，另一个就是Stream。新特性的加入，一定是为了某种需求，那么Stream是什么，它能帮助我们做什么？首先看下面这个例子：

有这样一份数据，一组考卷List，每个Paper有三个属性分别是学生名字studentName、课程名称className和分数score。现在我们需要从中找出语文不及格（分数低于60）的学生名字，并且按分数从高到低排序。在不使用Java8新特性之前，相信大部分人都是像下面这样写的：

public static List<String> getFailedPaperStudentNamesByJava7(List<Paper> papers) {
    // 筛选出不及格的考卷
    List<Paper> failedPapers = new ArrayList<>();
    for (Paper paper : papers) {
        if (paper.getClassName().equals("语文") 
            && paper.getScore() < 60) {
            failedPapers.add(paper);
        }
    }
    // 按分数从高到低排序
    Collections.sort(failedPapers, new Comparator<Paper>() {
        @Override
        public int compare(Paper o1, Paper o2) {
            return o2.getScore() - o1.getScore();
        }
    });
    // 记下不及格的学生名字
    List<String> failedPaperStudentNames = new ArrayList<>();
    for (Paper failedPaper : failedPapers) {
        failedPaperStudentNames.add(failedPaper.getStudentName());
    }
    return failedPaperStudentNames;
}

下面是用Java8的Lambda表达式+Stream改写的版本：

public static List<String> getFailedPaperStudentNamesByJava8(List<Paper> papers) {
    return papers.stream()
        .filter(p -> p.getClassName().equals("语文")
                && p.getScore() < 60)
        .sorted((p1, p2) -> p2.getScore() - p1.getScore())
        .map(Paper::getStudentName)
        .collect(Collectors.toList());
}

可直观的看出，代码量少了，只用了一行代码把所有操作链接起来了。我们再细看下，从这方法的名字上去理解，首先通过stream从List获得Stream，然后可以使用流式操作处理数据，先是filter筛选出语文课且不及格的考卷，接着sorted对分数排序，再是map获得每个Paper中的学生名字，最后collect把所有的名字收集成一个List。可看出，从语义上的理解也更为直观了，在筛选语文课不及格的试卷时，我们不是使用命令式写法（遍历，然后判断，再放到一个新的List里），而是类似SQL中的where条件，通过声明式写法直接给出数据需要符合的条件，便能得到需要的数据。

我们说了很久的“Stream”，到底什么是“Stream”，笔者从Stream API这个角度谈自己的理解：

Stream是Java提供的一个接口，该接口允许以声明式的写法处理数据，可以把操作链接起来，形成数据处理流水线，还能将数据处理任务并行化。

声明式和链接操作，前面的例子已经能看出。那什么是并行化，例如统计学生名字，我们可以将该任务划分成多个，交给多个CPU分别计算，最后再汇总结果，这一系列复杂操作，可交给Stream完成，如下：

public static List<String> getFailedPaperStudentNamesByJava8(List<Paper> papers) {
    return papers.parallelStream()
        .filter(p -> p.getClassName().equals("语文")
                && p.getScore() < 60)
        .sorted((p1, p2) -> p2.getScore() - p1.getScore())
        .map(Paper::getStudentName)
        .collect(Collectors.toList());
}

只是将stream()方法改为parallelStream()方法就能将该任务并行化，是不是十分简单。

通过以上介绍，想必已对Stream有了初步认识，下面开始系统学习Stream。

1. 流和集合

首先我们还是要弄清楚流和集合在概念上的区别。集合，例如List、Set、Tree等，是一种存储数据的数据结构。关于数据，是已经存在了的，我们只是通过一种数据结构将数据组织起来，便于某种方式读取或保持某种结构。流不同于集合的地方在于数据并非在使用前全部获得，而是在使用过程中按需获得。例如文件流，我们可以通过readline将文件一行一行的读取。还有视频流，我们可以边看边下载，不用等将所有数据下载完毕才能观看。

所以虽然我们都能从集合、流中获取数据，但数据产生的时间是有区别的，集合的数据是预先产生的，而流则是根据需要实时产生的。两者的特性也导致用途上的差异，集合侧重存储，流侧重计算。因此我们常听到的流式计算的叫法。

下面说下流和集合在使用上的区别。集合，可以随时取用，但流在创建后只能被使用一次，若重复消费，则会报错。

List<String> list = Arrays.asList("A", "B", "C");
Stream<String> stream = list.stream();
stream.forEach(System.out::print);
stream.forEach(System.out::print);
// ABC
// java.lang.IllegalStateException: stream has already been operated upon or closed

另外，集合在遍历时，就像我们前面描述的例子，只能通过程序员编写 for-each 这种显示的代码去迭代，这被称作外部迭代。而流在遍历时，例如map会对流中的每个元素进行处理，所以我们不需要写具体的迭代代码，而是交由Java内部完成，这被称作内部迭代。内部迭代的好处在于，它是一个黑盒。你需要的是迭代，那Java只要完成你的目标就行了。而至于如何迭代，则交由Java来完成，这就为优化提供了可能，优化的方向有两点，一是更优化的顺序来处理，二是将操作并行化，例如我们在学生的示例中，只是将stream改成parallelStream()，后续其他的map操作可以根据判断是并行流从而将任务并行化，而我们不用修改map操作。

下面我们来学习如何使用流。

2. 创建流

在对流进行操作之前，我们首先需要获得一个Stream对象，创建流有以下几种方式。

2.1 集合
Collection的默认方法stream()，可以由集合类创建流。

List<String> list = Arrays.asList("A", "B", "C");
Stream<String> stream = list.stream();

2.2 值
Stream的静态方法of(T... values)，通过显示值创建流，可接受任意数量的参数。

Stream<String> stream = Stream.of("A","B","C");

2.3 数组
Arrays的静态方法stream(T[] array)从数组创建流，接受一个数组参数。

String[] ss = new String[]{"A", "B", "C"};
Stream<String> stream = Arrays.stream(ss);
2.4 文件
NIO中有较多静态方法创建流，例如Files的静态方法lines(Path path)从返回指定文件中的各行构成的字符串流。
try (Stream<String> stream = Files.lines(Paths.get("data.txt"))) {
    stream.forEach(System.out::print);
} catch (IOException e) {
}

2.5 函数
Stream的静态iterate和generate可根据函数计算创建无限流。首先看下iterate，通常用于依次生成一系列值，其声明如下：

Stream<T> iterate(final T seed, final UnaryOperator<T> f)

seed为初始值，UnaryOperator<T>是Function<T,R>的子类，区别在于规定输入、输出都是T类型，下面看个示例：

Stream.iterate(0, n -> n + 1)
    .forEach(System.out::println);

该示例会根据初始值，然后通过函数计算依次得到下一个值，0、1、2......你可以试着运行下，发现根本停不下来，这就是刚刚说到的无限流。我们可以通过limit(n)来对无限流做限制。

Stream.iterate(0, n -> n + 1)
    .limit(10)
    .forEach(System.out::println);

接着是generate，不同于iterate依次根据上次计算的结果生成， 而是通过一个Supplier<T>实例提供新的值，其声明如下：

Stream<T> generate(Supplier<T> s)

例如，我们生成10个随机数：

Stream.generate(Math::random)
    .limit(10)
    .forEach(System.out::println);

2.6 数值流
你可能已经注意到上述介绍的Stream<T>使用了泛型，所以可以适用于任意引用类型，而对于原始类型则只能使用其包装类，例如：

Stream.of(1, 2, 3)
    .forEach(n -> {
        System.out.println(n.getClass()); // Integer
        int x = n * 2; // 需要拆箱
    });

Stream在处理原始类型上会由于装箱拆箱造成较大的性能损耗，所以Java8提供了三种特殊的流接口IntStream、DoubleStream、LongStream，将流中的元素特化为int、double和long。

IntStream.of(1, 2, 3)
    .forEach(n -> {
        int x = n * 2; // n为int
    });

除了使用of创建流，还可以将普通流转成数值流，mapToInt、mapToDouble和mapToLong。

Stream.of(1, 2, 3)
    .mapToInt(Integer::intValue)
    .forEach(n -> {
        int x = n * 2; // n为int
    });

数值流也能转成普通流，boxed装箱。

IntStream.of(1, 2, 3)
    .boxed()
    .forEach(n -> {
        int x = n * 2; // n为Integer
    });

3. 流的操作

流创建好了，下面学习对流进行操作。流的操作分为两种：

  • 中间操作：返回一个Stream对象，可以将一系列中间操作构成一条流的流水线（类似构造器模式）
  • 终端操作：执行流水线，返回不是流的结果（也可是void）
public static List<String> getFailedPaperStudentNamesByJava8(List<Paper> papers) {
    return papers.parallelStream()
        .filter(p -> p.getClassName().equals("语文")  // Stream<Paper>
                && p.getScore() < 60)
        .sorted((p1, p2) -> p2.getScore() - p1.getScore())  // Stream<Paper>
        .map(Paper::getStudentName)  // Stream<String>
        .collect(Collectors.toList()); // List<String>
}

这里的filter、sorted、map就是中间操作，collect为终端操作。终端操作用于执行流水线是什么意思？意思是如果没有终端操作，将不会执行前面链接的中间操作。例如：

List<String> list = Arrays.asList("A", "B", "C");
list.stream()
    .map(s -> {
        System.out.print(s);
        return s;
    });
// 无输出

无终端操作的情况下，中间操作map里的代码块将不会执行，不会有输出。而我们在此基础上加上一个终端操作forEach，便能触发流水线的执行。

List<String> list = Arrays.asList("A", "B", "C");
list.stream()
    .map(s -> {
        System.out.print(s);
        return s;
    })
    .forEach(s -> {});
// ABC

下面来看下常用的流操作。

3.1 筛选

（1）filter

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

过滤

  • 接受一个谓词Predicate（T -> boolean）
  • 返回一个包含所有符合谓词的元素的流。
List<String> list = Arrays.asList("AA", "AB", "BC");
list.stream()
    .filter(s -> s.startsWith("A"))
    .forEach(System.out::println);
// AA
// AB

（2）distinct

Stream<T> distinct()

去重

  • 根据流中元素的hashCode和equals方法比较元素
  • 返回一个元素各异的流
    List<String> list = Arrays.asList("A", "A", "B");
    list.stream()
    .distinct()
    .forEach(System.out::print);
    // AB

3.2 切片
（1）limit

Stream<T> limit(long maxSize);

截断

  • 接受一个长度
  • 返回一个不超过给定长度的流
List<String> list = Arrays.asList("A", "B", "C");
list.stream()
    .limit(2)
    .forEach(System.out::print);
// AB

（2）skip

Stream<T> skip(long n)

跳过元素

  • 指定跳过前n个元素
  • 如果元素不足n个，返回一个空流
List<String> list = Arrays.asList("A", "B", "C");
list.stream()
    .skip(2)
    .forEach(System.out::print);
// C

3.3 映射

（1）map

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

对每个元素应用函数

  • 接受一个函数（T -> R）
  • 将每一个元素映射成一个新的元素
List<Paper> papers = Arrays.asList(
    new Paper("小明", "语文", 40),
    new Paper("小红", "语文", 80),
    new Paper("小蓝", "语文", 50)
);
papers.stream()
    .map(Paper::getStudentName)
    .forEach(System.out::println);
// 小明
// 小红
// 小蓝

（2）flatMap

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

流的扁平化，什么意思，先看下面这个例子：

List<String> list = Arrays.asList("ABC", "DEF", "GHI");
list.stream()
    .map(s -> s.split("")) // Stream<String[]>
    .forEach(System.out::println); 
// [Ljava.lang.String;@2f4d3709
// [Ljava.lang.String;@4e50df2e
// [Ljava.lang.String;@1d81eb93

我们想将字符串“ABC”，“DEF”，“GHI”三个字符中的每个字符组合成一个流然后打印出来，但是上述的写法，通过split函数拆分了String[]数组，流中的元素也被映射成了数组，例如String[]{"A", "B", "C"}，所以forEach打印得到的结果是数组地址。

我们如何才能把数组中的元素组合在一起，得到"A", "B", "C", "D"...的一个流呢。这就需要扁平化的处理。flatMap接受一个函数（T -> Stream），把流中每个元素映射为一个流，然后再把所有的流组合成一个最终的流。例如这里的元素是String[]，那我们就把数组映射成流Arrays::stream，这样就能把每个数组里的元素连接在一起了。

List<String> list = Arrays.asList("ABC", "DEF", "GHI");
list.stream()
    .map(s -> s.split("")) // Stream<String[]>
    .flatMap(Arrays::stream) // Steam<String>
    .forEach(System.out::println);

3.4 匹配
（1）anyMatch

boolean anyMatch(Predicate<? super T> predicate)

至少匹配一个元素

  • 接受一个谓词（T -> boolean）
  • 如果有一个元素匹配，返回true，否则返回false
List<String> list = Arrays.asList("A", "B", "C");
list.stream()
    .map(s -> {
        System.out.print(s);
        return s;
    })
    .anyMatch(s -> s.startsWith("B")); // true
// AB

当遇到B时，匹配到了，便会直接返回，不会再迭代后续元素，这是一种短路操作。

（2）allMatch

boolean allMatch<Predicate<? super T> predicate>

匹配所有元素

  • 接受一个谓词（T -> boolean）
  • 所有所有元素匹配，返回true，否则返回false
  • 当有一个元素不匹配，就会短路返回
List<String> list = Arrays.asList("A", "B", "C");
list.stream()
    .map(s -> {
        System.out.println(s);
        return s;
    })
    .allMatch(s -> s.startsWith("B")); // false
// A

（3）nonMatch

boolean nonMatch(Predicate<? super T> predicate)

所有元素不匹配

  • 接受一个谓词（T -> boolean）
  • 所有元素匹配，返回true，否则返回false
  • 当有一个元素匹配，就会短路返回
List<String> list = Arrays.asList("A", "B", "C");
list.stream()
    .map(s -> {
        System.out.println(s);
        return s;
    })
    .noneMatch(s -> s.startsWith("B")); // false
// A

3.5 查找
（1）findAny

Optional<T> findAny()

返回当前流中的任意元素，用Optional封装元素，迫使显示检查元素是否存在。

（2）findFirst

Optional<T> findFirst()

返回当前流中的第一个元素

对比：

两者都是返回一个元素，如果不关心返回的元素是哪个，优先使用findAny，因为这样在并行上的限制更少，可优化的空间更大。

3.6 归约

reduce

// 有初始值
T reduce(T identity, BinaryOperator<T> accumulator)
// 无初始值
Optional<T> reduce(BinaryOperator<T> accumulator)

通过接收一个BinaryOperator(T, T) -> T，将两个元素结合产生一个新值。reduce将一直执行该操作直到最后流中只剩一个元素返回。

int[] nums = new int[]{1, 2, 3, 4, 5};
int sum = IntStream.of(nums).reduce(0, Integer::sum); // 15
int max = IntStream.of(nums).reduce(Integer::max).orElse(-1); // 5
int min = IntStream.of(nums).reduce(Integer::min).orElse(-1); // 1

3.7 数值流的特殊操作

在2.6节中我们说过针对原始类型有特殊的原始类型流，由于都是数值，所以也设计了些针对数值的方法。

int[] nums = new int[]{1, 2, 3, 4, 5};
int sum = IntStream.of(nums).sum(); // 15，等同reduce(0, Integer::sum)
int max = IntStream.of(nums).max().orElse(-1); // 5，等同reduce(Integer::max).orElse(-1)
int min = IntStream.of(nums).min().orElse(-1); // 1，等同reduce(Integer::min).orElse(-1)
double avg = IntStream.of(nums).average().orElse(-1); //3.0

3.8 收集
collect在前面的示例中已经见过了，可以将流中的元素进行汇总。

<R, A> R collect(Collector<? super T, A, R> collector);

接收一个Collector收集器。在Collectors中已经内置了一些常用的收集器。

  • toList()：将元素收集成一个List。
  • toSet()：将元素收集成一个Set。
  • counting()：统计元素数量。
  • maxBy(Comparator<? super T> comparator)：获取元素中的最大值。
  • minBy(Comparator<? super T> comparator)：获取元素中的最小值。
  • summingInt(ToIntFunction<? super T> mapper)：将元素映射成一个int值，然后求和，类似的还有double和long。
  • averagingInt(ToIntFunction<? super T> mapper)：将元素映射成一个int值，然后求平均。
  • summarizingInt(ToIntFunction<? super T> mapper)：将元素映射成一个int值，然后得到一个IntSummaryStatistics对象，包含了统计数、总和、最大值、最小值和平均值。
  • joining()：把元素toSting()的结果连接成一个字符串，还有一个重载版本，接收一个分隔符参数。
  • groupingBy(Function<? super T, ? extends K> classifier)：接收一个Function，返回一个Map<K, List<T>>。通过Function的返回值作为Key，然后将具有相同Key的元素，组合成List。

下面看示例：

List<Paper> papers = Arrays.asList(
    new Paper("小明", "语文", 40),
    new Paper("小明", "数学", 80),
    new Paper("小红", "语文", 80),
    new Paper("小红", "数学", 80),
    new Paper("小蓝", "语文", 50),
    new Paper("小蓝", "数学", 60)
);
// 所有语文卷子
List<Paper> chinesePapers = papers.stream()
    .filter(p -> p.getClassName().equals("语文"))
    .collect(toList());
// 所有学科
Set<String> classNames = papers.stream()
    .map(Paper::getClassName)
    .collect(toSet());
// 最高分的卷子，最低分改成minBy就行
Paper maxScorePaper = papers.stream()
    .collect(maxBy((p1, p2) -> p1.getScore() - p2.getScore())).get();
// 总分数
int sumScore = papers.stream()
    .collect(summingInt(Paper::getScore));
// 平均分
double avgScore = papers.stream()
    .collect(averagingInt(Paper::getScore));
// 统计数、总和、最大值、最小值和平均值
IntSummaryStatistics summaryStatistics = papers.stream()
    .collect(summarizingInt(Paper::getScore));
long count = summaryStatistics.getCount();
long sum = summaryStatistics.getSum();
int max = summaryStatistics.getMax();
int min = summaryStatistics.getMin();
double avg = summaryStatistics.getAverage();
// 学生名字连接在一起
String studentNameStr = papers.stream()
    .map(Paper::getStudentName)
    .distinct()
    .collect(joining(","));
// 按学科将卷子分组
Map<String, List<Paper>> groupPapers = papers.stream()
    .collect(groupingBy(Paper::getClassName));

以上介绍的是常用的Collector，我们还可以根据需要自定义Collector，本文就不叙述了。

4. 总结

本文列举了在日常开发中较为常用的流操作，但是还有未涉及之处，感兴趣的读者可以直接看Stream的API。本文也没有讲述并行流，虽然用法简单，但是能否真正提高效率，还是要看具体情况，还缺乏经验就不叙述了。先掌握基本的流式操作吧。

上一篇：SwiftUI 简明教程之 GeometryReader、PreferenceKey


下一篇：OSPF配置-2