Stream是从某个数据源获得的支持聚合操作的元素序列。

名词解释：

• 元素序列：针对特定元素类型的有序集合流提供了一个接口。流不会存储元素，只会按需计算。
• 数据源：流所用到的数据源来自集合、数组或者I/O。
• 聚合操作：类似SQL语句一样的操作，比如filter, map, reduce, find, match, sorted等。

流的创建

Java8在推出流的同时，对集合框架也进行了一些比较大变更。主要是在Collection接口上提供了两种生成Stream的方法：

• stream()方法：该方法以集合作为源，返回集合中的所有元素以在集合中出现的顺序组成的流。
• parallelStream()方法：该方法以集合作为源，返回一个支持并发操作的流。

流的操作

流的使用一般包括三件事:

• 一个数据源来执行一个查询;
• 一个中间操作链，形成一条流的流水线;
• 一个终端操作，执行流水线，并能生成结果。

中间操作

终端操作

流的使用

forEach

迭代流的每个元素。

dishList.stream().forEach(System.out::println);
1

sorted

对流中元素进行排序。

dishList.stream().sorted(Comparator.comparingInt(Dish::getCalories)).forEach(System.out::println);
1

filter

过滤元素。

List<Dish> filter = dishList.stream().filter(Dish::isVegetarian).collect(Collectors.toList());
1

distinct

去除重复元素。

List<Integer> distinct = Arrays.asList(1, 5, 6, 5, 8, 1, 9, 10).stream().distinct().collect(Collectors.toList());
1

limit

限制返回元素个数。

List<Integer> limit = Arrays.asList(1, 5, 6, 5, 8, 1, 9, 10).stream().distinct().limit(3).collect(Collectors.toList());
1

skip

跳过前面的元素个数，与limit联合使用可以实现类似sql的分页功能。

List<Integer> skip = Arrays.asList(1, 5, 6, 5, 8, 1, 9, 10).stream().distinct().skip(2).limit(3).collect(Collectors.toList());
1

map

可以将流转换成另外一个元素的流。

List<String> map = dishList.stream().map(Dish::getName).collect(Collectors.toList());
1

flatMap

流的扁平化（将流中的流扁平化为一个流）。

List<String> flatMap = dishList.stream().flatMap(d -> Arrays.stream(d.getName().split(""))).distinct().collect(Collectors.toList());
1

anyMatch

只要有一个元素匹配就返回true。

boolean anyMatch = Arrays.asList(1, 5, 6, 5, 8, 1, 9, 10).stream().anyMatch(i -> i % 2 == 0);
1

allMatch

所有元素匹配返回true。

boolean allMatch = Arrays.asList(1, 5, 6, 5, 8, 1, 9, 10).stream().allMatch(i -> i % 2 == 0);
1

findAny

找到任意一个元素就返回。

Arrays.asList(1, 5, 6, 5, 8, 1, 9, 10).stream().findAny().ifPresent(System.out::println);
1

findAny()操作，返回的元素是不确定的，对于同一个列表多次调用findAny()有可能会返回不同的值。使用findAny()是为了更高效的性能。如果是数据较少，串行地情况下，一般会返回第一个结果，如果是并行的情况，那就不能确保是第一个。

findFirst

返回第一个元素。

Arrays.asList(1, 5, 6, 5, 8, 1, 9, 10).stream().findFirst().ifPresent(System.out::println);
1

sum

对所有的元素求和。

Optional.ofNullable(dishList.stream().map(Dish::getCalories).reduce(0, Integer::sum)).ifPresent(System.out::println); // 等价于下面的，注意返回值不一样
dishList.stream().map(Dish::getCalories).reduce(Integer::sum).ifPresent(System.out::println);
1
2

max

对所有的元素求最大值。

dishList.stream().map(Dish::getCalories).reduce(Integer::max).ifPresent(System.out::println);
1

min

对所有的元素求最小值。

dishList.stream().map(Dish::getCalories).reduce(Integer::min).ifPresent(System.out::println);
1

stream中三个参数的reduce方法的理解

方法定义

    <U> U reduce(U identity,
                 BiFunction<U, ? super T, U> accumulator,
                 BinaryOperator<U> combiner);
1
2
3

参数解释：

• 标识：组合函数的标识值，累加器的初始值。
• 累加器：一个关联的、不干扰的、无状态的函数，用于将额外的元素合并到结果中。
• 组合器：用于组合两个值的关联、不干扰、无状态函数，必须与累加器函数兼容。

第三个参数用于在并行计算下合并各个线程的计算结果，并行流运行时，内部使用了fork-join框架。

多线程时，多个线程同时参与运算，多个线程执行任务，必然会产生多个结果，那么如何将他们进行正确的合并，这就是第三个参数的作用。

使用

package com.morris.java8.stream;

import java.util.stream.IntStream;

public class ReduceExample {

    public static void main(String[] args) {

        Integer sum = IntStream.rangeClosed(1, 1000).boxed().reduce(0, Integer::sum, (x, y) -> {
            // 不会执行，不影响结果
            return 0;
        });
        System.out.println("sum=" + sum);
        System.out.println("-------------");

        Integer sum2 = IntStream.rangeClosed(1, 1000).boxed().parallel().reduce(0, Integer::sum, (x, y) -> {
            System.out.print("thread name: " + Thread.currentThread().getName());
            System.out.print(" x=" + x);
            System.out.println(" y=" + y);
            return x + y;
        });

        System.out.println("sum2=" + sum2);

    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

运行结果如下:

sum=500500
-------------
thread name: main x=40703 y=45297
thread name: main x=32953 y=37422
thread name: main x=70375 y=86000
thread name: main x=56203 y=61047
thread name: main x=48453 y=53172
thread name: main x=101625 y=117250
thread name: main x=156375 y=218875
thread name: ForkJoinPool.commonPool-worker-2 x=17453 y=21672
thread name: ForkJoinPool.commonPool-worker-1thread name: ForkJoinPool.commonPool-worker-2 x=9703 x=25203 y=13797
 y=29547
thread name: ForkJoinPool.commonPool-worker-2 x=39125 y=54750
thread name: ForkJoinPool.commonPool-worker-1 x=1953 y=5922
thread name: ForkJoinPool.commonPool-worker-1 x=7875 y=23500
thread name: ForkJoinPool.commonPool-worker-1 x=31375 y=93875
thread name: ForkJoinPool.commonPool-worker-1 x=125250 y=375250
sum2=500500
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

从运行结果可知，reduce方法的第三个参数用于在并行计算下合并各个线程的计算结果。

数值流

在Stream里元素都是对象，那么，当我们操作一个数字流的时候就不得不考虑一个问题，拆箱和装箱。虽然自动拆箱不需要我们处理，但依旧有隐含的成本在里面。

Java8引入了3个原始类型特化流接口来解决这个问题：IntStream,DoubleStream,LongStream, 分别将流中的元素特化为int、long、double，从而避免了暗含的装箱成本。

对象流转数值流

将对象流转换为数值流的常用方法是mapToInt、mapToDouble和mapToLong。

IntStream intStream = Arrays.asList(1, 3, 4, 6, 9).stream().mapToInt(Integer::intValue);
1

数值流转对象流

使用boxed方法。

Stream<Integer> boxed = intStream.boxed();
1

数值范围

有时候需要生成一个数值范围，比如1到30. 可以直接使用数值流生成。

IntStream.range(1, 5).forEach(System.out::print); // 不包含结束值
System.out.println();
IntStream.rangeClosed(1, 5).forEach(System.out::print); // 包含结束值
System.out.println();
1
2
3
4

数值流特殊函数

// min
IntStream.rangeClosed(1, 5).min().ifPresent(System.out::println);

// max
IntStream.rangeClosed(1, 5).max().ifPresent(System.out::println);

// sum
System.out.println(IntStream.rangeClosed(1, 5).sum());

// summaryStatistics
System.out.println(IntStream.rangeClosed(1, 5).summaryStatistics());
1
2
3
4
5
6
7
8
9
10
11

默认值OptionalInt

Optional可以用Integer、String等参考类型来参数化。对于三种数据流，也分别有一个Optional原始类型：OptionalInt、OptionalDouble和OptionalLong。
IntStream.rangeClosed(1, 5).min()有可能没有最小值，这时返回的是一个OptionalInt，返回0就不合理了。

构建流

从值生成流

Optional.ofNullable(Stream.of("hello", "morris", "world", "stream").collect(Collectors.joining(","))).ifPresent(System.out::println);
1

从数组生成流

Optional.ofNullable(Arrays.stream(new int[]{2, 3, 1, 4}).boxed().map(i -> String.valueOf(i)).collect(Collectors.joining(","))).ifPresent(System.out::println);
1

从文件生成流

try(Stream<String> stream = Files.lines(Paths.get("D:\\gitPrj\\morris-book\\Java\\rocketmq\\java8\\src\\main\\java\\com\\morris\\java8\\stream\\Trader.java"))) {
    stream.limit(5).forEach(System.out::println);
} catch (IOException e) {
    e.printStackTrace();
}
1
2
3
4
5

无限流

使用无限流时注意使用limit限制流的大小，否则会一直无限生成下去。

// iterate生成流
Optional.ofNullable(Stream.iterate(0, n -> n + 2).limit(10).map(i -> String.valueOf(i)).collect(Collectors.joining(","))).ifPresent(System.out::println);

// generate生成流
Optional.ofNullable(Stream.generate(Math::random).limit(5).map(d -> String.valueOf(d)).collect(Collectors.joining(","))).ifPresent(System.out::println);
1
2
3
4
5