0%

Java 8 中的 Streams API

发表于 更新于 分类于 Disqus：
Stream 作为 Java 8 的一大亮点，它与 java.io 包里的 InputStream 和 OutputStream 是完全不同的概念。它也不同于 StAX 对 XML 解析的 Stream，也不是 Amazon Kinesis 对大数据实时处理的 Stream。

Stream 作为 Java 8 的一大亮点，它与 java.io 包里的 InputStream 和 OutputStream 是完全不同的概念。它也不同于 StAX 对 XML 解析的 Stream，也不是 Amazon Kinesis 对大数据实时处理的 Stream。Java 8 中的 Stream 是对集合（Collection）对象功能的增强，它专注于对集合对象进行各种非常便利、高效的聚合操作（aggregate operation），或者大批量数据操作 (bulk data operation)。Stream API 借助于同样新出现的 Lambda 表达式，极大的提高编程效率和程序可读性。同时它提供串行和并行两种模式进行汇聚操作，并发模式能够充分利用多核处理器的优势，使用 fork/join 并行方式来拆分任务和加速处理过程。通常编写并行代码很难而且容易出错, 但使用 Stream API 无需编写一行多线程的代码，就可以很方便地写出高性能的并发程序。所以说，Java 8 中首次出现的 java.util.stream 是一个函数式语言+多核时代综合影响的产物。
https://www.ibm.com/developerworks/cn/java/j-lo-java8streamapi/

以上是网络上对于Java 8 Stream Api意义的解释。简单来说，Stream Api 给了一种比较方便合适的方式来做聚合操作。 在 Java 中，聚合操作往往只能通过数据库和普通循环操作来实现，这种方法写起来笨拙而且不够优雅。 例如在 Java 8 之前，我们找出指定部门的员工，然后将工号以排序输出到另一个 List 中，就只能这么写：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
List<Employee> employeeList = new ArrayList<>();
for (Employee t : AllEmployeeList) {
    if (t.getDepartment().equals("机票Bu")){
        employeeList.add(t);
    }
}
Collections.sort(employeeList, new Comparator<Employee>() {
    @Override
    public int compare(Employee o1, Employee o2) {
        return o1.getSn() - o2.getSn();
    }
});
List<Integer> employeeIds = new ArrayList<>();
for (Employee t:employeeList){
    employeeIds.add(t.getSn());
}

但是在 Java 8 中，我们有了更简洁，更高效的方式，因为 Stream Api 还有并发的特性。

1
2
3
4
5
List<Integer> employeeIds = AllEmployeeList.parallelStream()
                .filter(f -> f.getDepartment().equals("机票Bu"))
                .sorted(Comparator.comparing(Employee::getSn))
                .map(Employee::getSn)
                .collect(Collectors.toList());

什么是流

Stream 不是集合元素，它不是数据结构并不保存数据，它是有关算法和计算的，它更像一个高级版本的 Iterator。原始版本的 Iterator，用户只能显式地一个一个遍历元素并对其执行某些操作；高级版本的 Stream，用户只要给出需要对其包含的元素执行什么操作，比如 “过滤掉长度大于 10 的字符串”、”获取每个字符串的首字母”等，Stream 会隐式地在内部进行遍历，做出相应的数据转换。

Stream 就如同一个迭代器（Iterator），单向，不可往复，数据只能遍历一次，遍历过一次后即用尽了，就好比流水从面前流过，一去不复返。

而和迭代器又不同的是，Stream 可以并行化操作，迭代器只能命令式地、串行化操作。顾名思义，当使用串行方式去遍历时，每个 item 读完后再读下一个 item。而使用并行去遍历时，数据会被分成多个段，其中每一个都在不同的线程中处理，然后将结果一起输出。Stream 的并行操作依赖于 Java7 中引入的 Fork/Join 框架（JSR166y）来拆分任务和加速处理过程。

流的操作

生成流

从集合创建流

  • stream() − 为集合创建串行流。
  • parallelStream() − 为集合创建并行流。
  • Stream.of(Collections) - 创建串行流
    其他构建方法
  • java.io.BufferedReader.lines() - 从 BufferedReader 构建流
  • java.util.stream.IntStream.range() - 指定 int 范围内循环的串行流
  • java.nio.file.Files.walk() - 指定路径遍历文件流
  • java.util.Spliterator
  • Random.ints()
  • BitSet.stream()
  • Pattern.splitAsStream(java.lang.CharSequence)
  • JarFile.stream()

流的方法

forEach

迭代每个流的元素的方法，如果对效率有要求可以在并行流的基础上进行遍历，此时元素顺序将无法保证按照原有顺序

1
allEmployeeList.parallelStream().forEach(System.out::println);

map

映射、转换每个元素的方法。具体来说，是将原流的每个元素进行转换，合成一个新的流的方法。

1
2
3
4
5
//映射所有部门
allEmployeeList.stream().map(Employee::getDepartment).distinct().count();
//获取数组对应的平方数
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7);
List<Integer> squaresList = numbers.stream().map(i -> i * i).distinct().collect(Collectors.toList());

filter

按照条件过滤元素，经过过滤的元素生成一个新 Stream

1
2
3

//统计所有部门为 "机票Bu" 的员工数
allEmployeeList.stream().filter(e -> e.getDepartment().equals("机票Bu")).count();

limit / skip

限制流输出的元素个数，取流前 n 个元素生成新流

1
2
3
List<Integer> numbersOnLimit = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
List<Integer> squaresListOnLimit = numbersOnLimit.stream().limit(3).collect(Collectors.toList());
// squaresListOnLimit : [1, 2, 3]

skip 方法与 limit 相反，将跳过前 n 个元素生成新流

1
2
3
List<Integer> numbersOnSkip = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
List<Integer> squaresListOnSkip = numbersOnLimit.stream().skip(3).collect(Collectors.toList());
// squaresListOnSkip : [4, 5, 6, 7, 8, 9, 10]

还有一个需要注意的是由于 Stream 是管道形式的，在排序后执行 limit/skip 方法并不能起到减少排序的次数的作用

sorted

对流进行排序生成新流，

1
2
3
4
5
6
7
8
//对对象的排序，分别演示了排列 int 类型和 String 类型的操作
List<Employee> sortedEmployeeList = allEmployeeList.stream().sorted(Comparator.comparingInt(Employee::getSn)).collect(Collectors.toList());
List<Employee> sortedDepartmentEmployeeList = allEmployeeList.stream().sorted(Comparator.comparing(Employee::getDepartment)).collect(Collectors.toList());

//对数字的排序
List<Integer> numbersOnSort = Arrays.asList(9, 8, 2, 6, 1, 7, 3, 5, 4, 10);
List<Integer> squaresListOnSort = numbersOnLimit.stream().sorted().collect(Collectors.toList());
// squaresListOnSort : [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

min / max

在流中取最大或最小的方法，复杂度为 O(n),所以效率是相对排序要高很多。

1
Employee maxSn = allEmployeeList.stream().max(Comparator.comparingInt(Employee::getSn)).get();

get

从具有唯一值的流中取出一个值，通常 min / max / limit 等操作后获取一个值/对象的方法

orElse

讲了get，就必须要知道 orElse，因为过滤操作结束后流中并不一定是永远有值的，所以我们会在orElse中赋予一个默认值进行后续程序操作。

1
2
int orElseTest = numbers.stream().filter(i -> i > 10).max(Comparator.naturalOrder()).orElse(-1);
//输出-1，因为原 List 中没有大于10的值

allMatch / anyMatch / noneMatch

  • allMatch：Stream 中全部元素符合条件，返回 true
  • anyMatch：Stream 中只要有任意元素符合条件，返回 true
  • noneMatch：Stream 中没有一个元素符合条件，返回 true 这个效率也会很高，因为在判断时候不需要遍历所有的元素，比如 allMatch 在有一个元素不符合条件时将直接返回
1
2
3
4
boolean anyMatch = numbers.stream().anyMatch(i -> i > 5);
boolean allMatch = numbers.stream().allMatch(i -> i > -1);
boolean noneMatch = numbers.stream().noneMatch(i -> i > 100);
//三个结果都为 true

reduce

reduce 可以有两个参数，可以说 sum, min, max, average 等都是特殊的reduce

1
2
3
4
5
6
7
8
9
10
11
12
// 字符串连接，concat = "ABCD"
String concat = Stream.of("A", "B", "C", "D").reduce("", String::concat); 
// 求最小值，minValue = -3.0
double minValue = Stream.of(-1.5, 1.0, -3.0, -2.0).reduce(Double.MAX_VALUE, Double::min); 
// 求和，sumValue = 10, 有起始值
int sumValue = Stream.of(1, 2, 3, 4).reduce(0, Integer::sum);
// 求和，sumValue = 10, 无起始值
sumValue = Stream.of(1, 2, 3, 4).reduce(Integer::sum).get();
// 过滤，字符串连接，concat = "ace"
concat = Stream.of("a", "B", "c", "D", "e", "F").
 filter(x -> x.compareTo("Z") > 0).
 reduce("", String::concat);