Java 8 Stream 从入门到进阶——像SQL一样玩转集合

0.阅读完本文你将会

• 了解Stream的定义和它的特征
• 了解Stream的基础和高阶用法

1. 前言

在我们日常使用Java的过程中，免不了要和集合打交道。对于集合的各种操作有点类似于SQL——增删改查以及聚合操作，但是其方便性却不如SQL。

所以有没有这样一种方式可以让我们不再使用一遍又一遍的循环去处理集合，而是能够便捷地操作集合？

答案是有的，它就是——Java 8引入的Stream，亦称为流 。

2. 流的定义

A Stream is a sequence of elements from a source.

流是一个来自数据源的元素队列。

简单来说，流是对数据源的包装，它允许我们对数据源进行聚合操作，并且可以方便快捷地进行批量处理。

日常生活中，我们看见水流在管道中流淌。Java中的流也是可以在“管道”中传输的。并且可以在“管道”的节点进行处理，比如筛选，排序等。

+--------------------+       +------+   +------+   +---+   +-------+
| stream of elements +-----> |filter+-> |sorted+-> |map+-> |collect|
+--------------------+       +------+   +------+   +---+   +-------+

元素流在管道中经过中间操作（intermediate opertaion）的处理，最后由终端操作（terminal opertaion）得到前面处理的结果（每一个流只能有一次终端处理）。

中间操作可以分为无状态操作和有状态操作，前者是指元素的处理不受之前元素的影响；后者是指该操作只有拿到所有元素才能继续下去。

终端操作也可分为短路与非短路操作，前者是指遇到符合条件的元素就可以得到最终结果，而后者必须处理所有元素才能得到最终结果。

下图为我们展示了中间操作和终端操作的具体方法。

如何快速区分中间操作和终端操作？

看方法的返回值，返回值为Stream的一般都是中间操作，否则是终端操作。

再来看看流的特征：

1. 流并不存储数据，所以它不是一个数据结构，它也不会修改底层的数据源，它为了函数式编程而生。

2. 惰性执行的，例如filter,map等都是延迟执行的。流的中间操作总是惰性的。

当终端操作需要中间操作时，中间操作才会被调用。

我们来看一个说明这点的例子：

List<String> list = Arrays.asList("a", "b", "c");
Stream<String> stream = list.stream().filter(element -> {
    System.out.println("filter() was invoked");
    return element.contains("b");
    });

上面这段代码告诉我们，流的元素有三个，所以我们应该是调用三次filter() 方法，应该打印三次filter() was invoked。

但实际上一次也没有打印，这就说明其实filter() 方法没有调用过一次。这是因为代码中缺失了终端操作。

我们改动下代码，添加一个map()方法以及终端操作。

List<String> list = Arrays.asList("a", "b", "c");
Optional<String> stream = list.stream().filter(element -> {
      System.out.println("filter() was invoked");
      return element.contains("b");
}).map(ele -> {
      System.out.println("map() was invoked");
      return ele.toUpperCase();
}).findFirst();

输入结果如下：

filter() was invoked
filter() was invoked
map() was invoked

打印结果显示，我们调用了两次filter() 方法，一次map() 方法。

在上面这段代码中，流的第一个元素不符合filter的条件，然后第二次调用，找到了符合的元素，接下来程序没有第三次调用filter()方法，而是"顺着管道"直接调用了map() 方法。

因为findFirst() 方法只选取第一个元素，所以我们至少少调用了一个filter() 方法。这正是因为惰性调用的机制。

1. 流有可能是无限的。虽然集合具有有限的大小，但流不需要。短路操作，如limit(n)或findFirst()，允许在有限的时间内完成对无限流的计算。

2. 流还是消耗品。在流的生命周期中，流的元素只被访问一次。与迭代器一样，必须生成新的流来重新访问源的相同元素。被访问过的流会被关闭。

针对第4点，我们看一个例子。

IntStream intStream = IntStream.of(1, 2, 3);
OptionalInt anyElement = intStream.findAny();
OptionalInt firstElement = intStream.findFirst();

执行这段代码将会得到以下的报错java.lang.IllegalStateException：

Exception in thread "main" java.lang.IllegalStateException: stream has already been operated upon or closed

因为代码中的intStream已经经历过一次了终端操作findAny(),所以intStream已经被关闭，再执行一次终端操作就会报错。

这种设计是符合逻辑和流的特性，因为流并不是为了存储元素。

我们将代码改成下面这样的，就可以执行多次终端操作了。

int[] intArray = {1, 2, 3};
OptionalInt anyElement = Arrays.stream(intArray).findAny();
OptionalInt firstElement = Arrays.stream(intArray).findFirst();

以上这些特征将Stream与Collection区分开来。

请注意，这里的Stream“流”与Java I/O流是不同的。它们之间的关系很小。

3. 创建一个流

创建一个Java流有许多方式。一旦流被创建了，那么它是无法修改数据源的，所以针对一个数据源我们可以创建多个流。

3.1 创建一个空的流

我们可以使用empty() 方法来创建一个空的流：

Stream<String> emptyStream = Stream.empty();

我们还可以用empty() 方法来返回一个空流从而避免返回null：

public Stream<String> streamOf(List<String> list) {
    return list == null || list.isEmpty() ? Stream.empty() : list.stream();
}

3.2 使用数组创建流

我们可以使用数组的全部或者一部分来创建流：

String[] arr = new String[]{"1", "2", "3"，"4", "5"};
Stream<String> entireArrayStream = Arrays.stream(arr);
Stream<String> partArrayStream = Arrays.stream(arr, 1, 4);

3.3 使用集合创建流

我们也可以使用集合来创建流：

Collection<String> collection = Arrays.asList("1", "2", "3");
Stream<String> collectionStream = collection.stream();

3.4 使用Stream.Builder()来创建流

使用这种方式创建流的时候请注意，一定要声明好你想要的类型，否则创建的会是Stream<Obejct>的流：

Stream<String> streamBuilder =
  Stream.<String>builder().add("1").add("2").add("3").build();

3.5 使用File来创建流

我们可以通过Files.lines()方法来创建流。文件的每一行都会成为流的每一个元素。

Path path = Paths.get("C:\\tmp\\file.txt");
        Stream<String> fileStream = Files.lines(path);
        Stream<String> fileStreamWithCharset = Files.lines(path, Charset.forName("UTF-8"));

3.6 Stream.iterate()

我们还可以使用iterate() 来创建一个流：

Stream<Integer> iteratedStream = Stream.iterate(10, n -> n + 1).limit(10);

在上面这段代码中，将会创造一个连续元素的流。

第1个元素是10，第2个元素是11，依此类推，直到元素数量达到size。

3.7 Stream.generate()

generate() 方法接受一个Supplier<T>来生成元素。

因为流是无限的，所以我们需要设置流的size。

下面这段代码将会创建一个流，它包含了5个"ele"字符串。

Stream<String> generatedStream =
  Stream.generate(() -> "ele").limit(5);

3.8 基本类型的流

1. range()和rangeClosed()

在Java8中，三种基本类型——int,long,double可以创建对应的流。

因为Stream<T>是泛型接口，所以无法用基本类型作为类型参数，因为我们使用IntStream，LongStream，DoubleStream来创建流。

IntStream intStream = IntStream.range(1, 3);//1,2
LongStream longStream = LongStream.rangeClosed(1, 3);//1,2,3

range(int start, int end) 方法会创建一个从start到end的有序流，它的步长是1，但是它不包括end。

rangeClosed(int start, int end) 与range() 方法的区别在于，前者会包括end。

2. of()方法

此外，基本类型还可以通过of() 方法来创建流。

int[] intArray = {1,2,3};
IntStream intStream = IntStream.of(intArray);//1,2,3
IntStream intStream2 = IntStream.of(1, 2, 3);//1,2,3

long[] longArray = {1L, 2L, 3L};
LongStream longStream = LongStream.of(longArray);//1,2,3
LongStream longStream2 = LongStream.of(1L, 2L, 3L);//1,2,3

double[] doubleArray = {1.0, 2.0, 3.0};
DoubleStream doubleStream = DoubleStream.of(doubleArray);
DoubleStream doubleStream2 = DoubleStream.of(1.0, 2.0, 3.0);//1.0,2.0,3.0

3. Random类

另外，从Java8开始，Random类也提供了一系列的方法来生成基本类型的流。例如：

Random random = new Random();
IntStream intStream = random.ints(3);
LongStream longStream = random.longs(3);
DoubleStream doubleStream = random.doubles(3);

3.9 字符串的流

1. 字符的流

因为Java没有CharStream，所以我们用InStream来替代字符的流。

IntStream charStream = "abc".chars();

2. 字符串的流

我们可以通过正则表达式来创建一个字符串的流。

Stream<String> stringStream = Pattern.compile(",").splitAsStream("a,b,c");

4. 流的用法

4.1 基本用法

4.1.1 forEach()方法

我们对forEach() 方法应该很熟悉了，在Collection中就有。它的作用是对每个元素执行指定的动作，也就是对元素进行遍历。

Arrays.asList("Try", "It", "Now")
                .stream()
                .forEach(System.out::println);

输出结果：

Try
It
Now

1. 方法引用

可能会有读者疑惑System.out::println是什么写法，正常的写法不应该都是下面这样嘛？

Arrays.asList("Try", "It", "Now")
                .stream()
                .forEach(ele -> System.out.println(ele));

其实两者写法是等价的，只不过前者是后者的简写方式。前者这种语法形式叫做方法引用(method references)，这种语法用来替代某些特定形式的lambda表达式。

如果lambda表达式的全部内容就是调用一个已有方法，那么可以用方法引用来替代lambda表达式。

这一点很重要，也很值得学习，在下面的内容中也会有很多这样的简写。

我们插个题外话，我们可以将方法引用细分为以下四类：

而System.out::println就是引用了某个对象的方法。

2. 副作用

其实在上面这个例子中，我们使用forEach() 将结果打印出来是一个常见的使用副作用(Side-effects)的场景。

但是除了这场景之外，我们应该避免使用流的副作用。

按照我自己的理解就是，不要去修改函数外部的状态，不要在中间操作中对lambda表达式之外的属性产生写操作。

特别是在并行流里，这种操作会导致结果无法预测，因为并行流是无序的。

// 错误
List<String> list = new ArrayList<>();
stream.filter(s -> pattern.matcher(s).matches())
      .forEach(s -> list.add(s));//错误的副作用使用场景
// 正确
List<String> list2 =
     stream.filter(s -> pattern.matcher(s).matches())
             .collect(Collectors.toList());//无副作用

4.1.2 filter()方法

filter() 方法的作用是返回符合条件的Stream。

Arrays.asList("Try", "It", "Now")
                .stream()
                .filter(ele -> ele.length() == 3)
                .forEach(System.out::println);

输出结果：

Try
Now

4.1.3 distinct()方法

distinct() 方法返回一个去重的stream。

Arrays.asList("Try", "It", "Now", "Now")
                .stream()
                .distinct()
                .forEach(System.out::println);

4.1.4 sorted()方法

排序函数有两个，一个是自然顺序，还有一个是自定义比较器排序。

Arrays.asList("Try", "It", "Now")
                .stream()
                .sorted((str1, str2) -> str1.length() - str2.length())
                .forEach(System.out::println);

输出结果：

It
Try
Now

4.1.5 map()方法

map() 方法对每个元素按照某种操作进行转换，转换后流的元素不会改变，但是元素类型取决于转换之后的类型。

Arrays.asList("Try", "It", "Now")
                .stream()
                .map(String::toUpperCase)

输出结果：

TRY
IT
NOW

4.1.6 flatMap()方法

flat的英文就是”平坦的“意思，而flatMap()方法的作用就是将流的元素摊平，借助下面这个例子我们更好理解：

Stream.of(Arrays.asList("Try", "It"), Arrays.asList("Now"))
                .flatMap(list -> list.stream())
                .forEach(System.out::println);

输出结果：

Try
It
Now

在上述这段代码中，原来的stream有两个元素，分别是两个List，执行了flatMap()之后，将每个List都”摊平“成了一个个的元素，所以会产生一个有三个字符串组成的流。

4.2 归约操作

上一小节介绍了Stream的基本用法，但是如此强大的流又怎么能止步于此呢？ 下面让我们看看流的重头戏——归约操作。

归约操作（reduction operation）也被称为折叠操作(fold)，是通过某种连接动作将所有元素汇总成一个结果的过程。元素求和、求最大值、求最小值、求总数，将所有元素转换成一个集合等都属于归约操作。

Stream类库有两个通用的归约操作reduce()和collect() ，也有一些为简化书写而设计的专用归约操作，比如sum()、max()、min()、count()等。

这些都比较好理解，所以我们会重点介绍reduce()和collect()。

4.2.1 reduce()

reduce操作可以实现从一组元素中生成一个值，比如sum()、max()、min()、count()等都是reduce操作。

reduce()方法定义有三种形式：

Optional<T> reduce(BinaryOperator<T> accumulator)

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

<U> U reduce(U identity, BiFunction<U,? super T,U> accumulator, BinaryOperator<U> combiner)

1. identity-初始值

2. accumulator-累加器

3. combiner-拼接器，只有并行执行时才会用到。

让我们看看这三个方法的实例：

Optional<Integer> reducedInt = Stream.of(1, 2, 3).reduce((a, b) -> a + b);

reducedInt = 1 + 2 + 3 = 6 上面这段代码中没有初始值，只有累加器，那么就是很简单的a与b的累加。

int reduceIntWithTwoParams = Stream.of(1, 2, 3).reduce(10, (a, b) -> a + b);

reduceIntWithTwoParams = 10 + 1 + 2 + 3 = 16

上面这段代码有初始值和累加器，所以运算的时候先要加上初始值，然后再逐步累加。

int reducedIntWithAllParams = Stream.of(1, 2, 3).reduce(10, (a, b) -> a + b, (a, b) -> {
    System.out.println("Combiner was invoked.");
    return a + b;
});

这段代码的结果与上一段的结果相同，并且没有打印，说明combiner并没有被调用。如果需要使combiner起作用，我们在这里应该使用parallelStream()方法。

int reducedIntWithAllParams = Arrays.asList(1, 2, 3).parallelStream().reduce(10, (a, b) -> a + b, (a, b) -> {
    System.out.println("Combiner was invoked");
    return a + b;
});

reducedIntWithAllParams = (10 + 1)+ ((10 + 2) + (10 + 3)) = 36

为什么是36呢？这是因为combiner的作用，它把多个并行结果拼接在了一起。

Collection.stream() 和 Collection.parallelStream() 分别产生序列化流(普通流)和并行流。

并行(parallel)和并发(concurrency)是有区别的。

并发是指一个处理器同时处理多个任务。而并行是指多个处理器或者是多核的处理器同时处理多个不同的任务。

并发是逻辑上的同时发生，而并行是物理上的同时发生。

打个比方：并发是一个人同时吃三个馒头，而并行是三个人同时吃三个馒头。

并且并行不一定快，尤其在数据量很小的情况下，可能比普通流更慢。只有在大数据量和多核的情况下才考虑并行流。

在并行处理情况下，传入给reduce()的集合类，需要是线程安全的，否则执行结果会与预期结果不一样。

4.2.2 collect()方法

collect()应该算是Stream里的最终王牌选手了，基本上你想要的功能都能在这里找到。

而且使用它也是Java函数式编程入门一个绝好的途径。

下面让我们从实际的例子出发吧！

List<Student> students = Arrays.asList(new Student("Jack", 90)
                , new Student("Tom", 85)
                , new Student("Mike", 80));

1. 常规归约操作

• 获取平均值

Double averagingScore = students.stream().collect(Collectors.averagingDouble(Student::getScore));

• 获取和

Double summingScore = students.stream().collect(Collectors.summingDouble(Student::getScore));

• 获取分析数据

DoubleSummaryStatistics doubleSummaryStatistics = students.stream().collect(Collectors.summarizingDouble(Student::getScore));

你可以从doubleSummaryStatistics 获取最大值、最小值、平均值等常见统计数据。

Collectors提供的这些方法省去了额外的map() 方法，当然你也可以先使用map() 方法，再进行操作。

2. 将流转换成Collection

通过以下的代码我们可以提取集合中的Student的Name属性，并且装入字符串类型的集合当中。

List<String> studentNameList = students.stream().map(Student::getName).collect(Collectors.toList());//[Jack, Tom, Mike]

还可以通过Collectors.joining() 方法来连接字符串。 并且Collector会帮你处理后最后一个元素不应该再加分隔符的问题。

String studentNameList = students.stream().map(Student::getName).collect(Collectors.joining(",", "[", "]"));//打印出来就是[Jack,Tom,Mike]

3. 将流转换成Map

Map不能直接转换成Stream，但是Stream生成Map是可行的，在生成Map之前，我们应该先定义好Map的Key和Value分别代表什么。

通常在下面三种情况下collect()的结果会是Map:

• Collectors.toMap(),使用者需要指定Map的key和value；
• Collectors.groupingBy(),对元素进行group操作；
• Collectors.partitioningBy()，对元素进行二分区操作。

Collectors.toMap()

下面这个例子为我们展示了怎么将students列表转换成<Student student, double score>组成的map。

Map<Student, Double> collect = students
                .stream()
                .collect(Collectors.toMap(Function.identity(), Student::getScore));

Collectors.groupingBy()

这个操作有点类似于SQL中的groupBy操作，按照某个属性对数据进行分组，而属性相同的元素会被分配到同一个key上。

而下面这个例子将会把Student按照Score进行分组：

Map<Double, List<Student>> nameStudentMap = students.stream().collect(Collectors.groupingBy(Student::getScore));

Collectors.partitioningBy()

partitioningBy()按照某个二元逻辑将stream中的元素分为两个部分，比如说下面这个例子将Student分成了成绩及格或者不及格的部分。

Map<Boolean, List<Student>> map = students.stream().collect(Collectors.partitioningBy(ele -> ele.getScore() >= 85));

打印结果：

{false=[Student{name='Mike', score=80.0}], true=[Student{name='Jack', score=90.0}, Student{name='Tom', score=85.0}]}

5. 结语

Java 8 Stream是一个强大的工具，但是我们在使用它的时候一定要符合规范，不然它可能会给你带来意想不到的惊喜哦~