Stream 概述
- Stream 是一个可以对序列中的每个元素执行流操作的一个元素序列
- Stream 包含中间和最终两种形式的操作，中间操作返回的还是一个 Stream，因此可以使用链式调用，最终操作返回的是 void 或一个非 Stream 的结果
- 下面代码中：filter、map、sorted 是中间操作，forEach 是最终操作
List<String> myList =
Arrays.asList("a1", "a2", "b1", "c2", "c1");
myList
.stream()
.filter(s -> s.startsWith("c"))
.map(String::toUpperCase)
.sorted()
.forEach(System.out::println);
// C1
// C2
- java Collection 接口中新增了 stream() 方法和 parallelStream() 方法，这些方法可以创建一个顺序的 stream 或者并发的 stream（parallelStream），并发的 stream 适合在多线程中使用
Stream 创建
- List 对象调用 stream() 方法返回一个对象流
Arrays.asList("a1", "a2", "a3")
.stream()
.findFirst()
.ifPresent(System.out::println); // a1
- 也可以使用 Stream.of 方法来创建 Stream
Stream.of("a1", "a2", "a3")
.findFirst()
.ifPresent(System.out::println); // a1
- IntStream、LongStream、DoubleStream 这些流能够处理基本数据类型如：int、long、double
- 比如：IntStream 可以使用 range() 方法能够替换掉传统的 for 循环
IntStream.range(1, 4)
.forEach(System.out::println);
// 1
// 2
// 3
- 基本类型流比常规对象流类型多了许多聚合方法，如：sum()、average()等
Arrays.stream(new int[] {1, 2, 3})
.map(n -> 2 * n + 1)
.average()
.ifPresent(System.out::println); // 5.0
- 可以通过常规对象流的 mapToInt() 、mapToLong()、mapToDouble() 方法完成基本类型之间的相互转化
IntStream.range(1, 4)
.mapToObj(i -> "a" + i)
.forEach(System.out::println);
- 下面例子 double Stream 先被映射成 int stream，然后又被映射成 String 类型对象流
Stream.of(1.0, 2.0, 3.0)
.mapToInt(Double::intValue)
.mapToObj(i -> "a" + i)
.forEach(System.out::println);
// a1
// a2
// a3
处理顺序
- Laziness（延迟加载）是中间操作的一个重要特性，只有最终操作时中间操作才会被执行，如下面例子
Stream.of("d2", "a2", "b1", "b3", "c")
.filter(s -> {
System.out.println("filter: " + s);
return true;
});
- 上面代码需要最终操作 forEach 才会执行
Stream.of("d2", "a2", "b1", "b3", "c")
.filter(s -> {
System.out.println("filter: " + s);
return true;
})
.forEach(s -> System.out.println("forEach: " + s));
//执行结果如下：
//filter: d2
//forEach: d2
//filter: a2
//forEach: a2
//filter: b1
//forEach: b1
//filter: b3
//forEach: b3
//filter: c
//forEach: c
- 每一个元素沿着链条垂直移动，第一个字符串 "d2" 执行完 filter 和 forEach 后才第二个元素，下面代码中，anyMatch 匹配到 A2 时立马返回为真，所以不会继续验证下一个元素 "b1"
Stream.of("d2", "a2", "b1", "b3", "c")
.map(s -> {
System.out.println("map: " + s);
return s.toUpperCase();
})
.anyMatch(s -> {最终操作
System.out.println("anyMatch: " + s);
return s.startsWith("A");
});
// map: d2
// anyMatch: D2
// map: a2
// anyMatch: A2
执行效率与 Stream 执行链顺序的关系
- 下面代码由两个中间操作 map 和 filter 以及一个最终操作 forEach 构成，我们观察这些动作是如何执行的
Stream.of("d2", "a2", "b1", "b3", "c")
.map(s -> {
System.out.println("map: " + s);
return s.toUpperCase();
})
.filter(s -> {
System.out.println("filter: " + s);
return s.startsWith("A");
})
.forEach(s -> System.out.println("forEach: " + s));
// map: d2
// filter: D2
// map: a2
// filter: A2
// forEach: A2
// map: b1
// filter: B1
// map: b3
// filter: B3
// map: c
// filter: C
- 上面代码：map 和 filter 被执行了 5 次，但 forEach 只执行了一次，可以修改操作顺序（如：将 filter 操作移动到操作链头部）来降低执行次数
- 修改后 map 只执行了 1 次，在数据量较大时会有很大的效率提升
Stream.of("d2", "a2", "b1", "b3", "c")
.filter(s -> {
System.out.println("filter: " + s);
return s.startsWith("a");
})
.map(s -> {
System.out.println("map: " + s);
return s.toUpperCase();
})
.forEach(s -> System.out.println("forEach: " + s));
// filter: d2
// filter: a2
// map: a2
// forEach: A2
// filter: b1中间操作
// filter: b3
// filter: c
- 有一些特殊的中间操作不会按流顺序执行，例如 sort 操作，它是一种特殊的中间操作，在对集合元素进行排序的过程中需要保存元素状态，是一种有状态的操作（stateful operation），在执行排序操作时（sorting 操作对集合进行水平操作），上面例子中 sorted 执行了8次，可以通过对链重排序方式，提升 stream 的执行效率
Stream.of("d2", "a2", "b1", "b3", "c")
.sorted((s1, s2) -> {
System.out.printf("sort: %s; %s\n", s1, s2);
return s1.compareTo(s2);
})
.filter(s -> {
System.out.println("filter: " + s);
return s.startsWith("a");
})
.map(s -> {
System.out.println("map: " + s);
return s.toUpperCase();
})
.forEach(s -> System.out.println("forEach: " + s));
//执行结果如下：
//sort: a2; d2
//sort: b1; a2
//sort: b1; d2
//sort: b1; a2
//sort: b3; b1
//sort: b3; d2
//sort: c; b3
//sort: c; d2
//filter: a2
//map: a2
//forEach: A2
//filter: b1
//filter: b3
//filter: c
//filter: d2
- 下面代码修改链顺序后由于有 filter 操作的过滤，导致 sorted 只有一个输入集元素，在大量数据下能提高效率
Stream.of("d2", "a2", "b1", "b3", "c")
.filter(s -> {
System.out.println("filter: " + s);
return s.startsWith("a");
})
.sorted((s1, s2) -> {
System.out.printf("sort: %s; %s\n", s1, s2);
return s1.compareTo(s2);
})
.map(s -> {
System.out.println("map: " + s);
return s.toUpperCase();
})
.forEach(s -> System.out.println("forEach: " + s));
// filter: d2
// filter: a2
// filter: b1
// filter: b3
// filter: c
// map: a2
// forEach: A2
流复用
- Stream 当你执行完任何一个最终操作时，流就被关闭了
Stream<String> stream =
Stream.of("d2", "a2", "b1", "b3", "c")
.filter(s -> s.startsWith("a"));
stream.anyMatch(s -> true); // ok
stream.noneMatch(s -> true); // exception
- 可以通过 Lambda 表达式创建 Supplier 接口实例，通过 get 方法创建一个 Stream，来实现流复用
Supplier<Stream<String>> streamSupplier =
() -> Stream.of("d2", "a2", "b1", "b3", "c")
.filter(s -> s.startsWith("a"));
streamSupplier.get().anyMatch(s -> true); // ok
streamSupplier.get().noneMatch(s -> true); // ok
Collect（收集）
- 假设存在如下用户类
class Person {
String name;
int age;
Person(String name, int age) {
this.name = name;
this.age = age;
}
@Override
public String toString() {
return name;
}
}
List<Person> persons =
Arrays.asList(
new Person("Max", 18),
new Person("Peter", 23),
new Person("Pamela", 23),
new Person("David", 12));
- Collect 可以把 Stream 中的元素转换为另一种形式，如：list、set、map
- Collect 使用 Collector 作为参数，Collector 包含四种不同操作：
- supplier（初始构造器）
- accaumulator（累加器）
- combiner（组合器）
- finisher（终结者）
- 这听起来很复杂，Java8 通过 Collectors 类内置了各种复杂的收集操作，因此对于大部分操作来说不需要自己去实现 Collector 类
- 例如下面代码将 Stream 转化为 List，如果想转换为 Set 的话，只需要用 Collectors.toSet() 就可以了
Map<Integer, List<Person>> personsByAge = persons
.stream()
.collect(Collectors.groupingBy(p -> p.age));
personsByAge
.forEach((age, p) -> System.out.format("age %s: %s\n", age, p));
// age 18: [Max]
// age 23: [Peter, Pamela]
// age 12: [David]
- 下面例子将用户按年龄分组
Map<Integer, List<Person>> personsByAge = persons
.stream()
.collect(Collectors.groupingBy(p -> p.age));
personsByAge
.forEach((age, p) -> System.out.format("age %s: %s\n", age, p));
// age 18: [Max]
// age 23: [Peter, Pamela]
// age 12: [David]
- Collectors 类功能繁多，你可以通过 Collectors 对 Stream 元素进行汇聚，如计算平均年龄
Double averageAge = persons
.stream()
.collect(Collectors.averagingInt(p -> p.age));
System.out.println(averageAge); // 19.0
- 可以通过 summarizing collectors 返回一个内置的统计对象，通过这个对象获取更全面的统计信息，比如用户年级中最大值、最小值、平均值等
IntSummaryStatistics ageSummary =
persons
.stream()
.collect(Collectors.summarizingInt(p -> p.age));
System.out.println(ageSummary);
// IntSummaryStatistics{count=4, sum=76, min=12, average=19.000000, max=23}
- 或者将用户名连接成一个字符串，join collector三个参数分别表示连接符、字符串前缀、字符串后缀
String phrase = persons
.stream()
.filter(p -> p.age >= 18)
.map(p -> p.name)
.collect(Collectors.joining(" and ", "In Germany ", " are of legal age."));
System.out.println(phrase);
// In Germany Max and Peter and Pamela are of legal age.
- 可以将 Stream 转换为 map，必须指定 map 的 key 和 value 如何映射，需要注意的是 key 的值必须是唯一的，否则会抛出 IllegalStateException，但可通过合并函数绕过这个异常
Map<Integer, String> map = persons
.stream()
.collect(Collectors.toMap(
p -> p.age,
p -> p.name,
(name1, name2) -> name1 + ";" + name2));
System.out.println(map);
// {18=Max, 23=Peter;Pamela, 12=David}
- 自定义 Collector，实现的功能是将 Stream 中所有用户的用户名变成大写并用 “|” 符号连接成一个字符串，可以通过 Collector.of() 创建一个 collector
- 由于 String 是不可变对象，所以需要一个辅助类 StringJoiner 帮助 collect 构造字符串，supplier 创建了一个包含分隔符的 StringJoiner 对象，accumulator 将用户名转换为大写并添加到 supplier 创建的 StringJoiner 中，combiner 将两个 StringJoiners 对象连接成一个，最后一步的 finisher 从 StringJoiner 中构建出所希望的 String 对象
Collector<Person, StringJoiner, String> personNameCollector =
Collector.of(
() -> new StringJoiner(" | "), // supplier
(j, p) -> j.add(p.name.toUpperCase()), // accumulator
(j1, j2) -> j1.merge(j2), // combiner
StringJoiner::toString); // finisher
String names = persons
.stream()
.collect(personNameCollector);
System.out.println(names); // MAX | PETER | PAMELA | DAVID
FlatMap
- 虽然我们可以使用 map 方法将 stream 中一种对象转化成另外一种对象，但 map 使用场景有限制，只能将一种对象映射为另外一种已存在的对象，是否能将一个对象映射为多种对象，或者映射成一个根本不存在的对象呢？这就是 flatMap 方法的作用
- FlatMap 方法可以将一个 Stream 中的每个元素转换为另一个 Stream 中的另一种元素，因此可以将 Stream 种每个对象改造成零、一个或多个
- 假设有以下类
class Foo {
String name;
List<Bar> bars = new ArrayList<>();
Foo(String name) {
this.name = name;
}
}
class Bar {
String name;
Bar(String name) {
this.name = name;
}
}
- 通过流实例化一队对象
List<Foo> foos = new ArrayList<>();
// create foos
IntStream
.range(1, 4)
.forEach(i -> foos.add(new Foo("Foo" + i)));
// create bars
foos.forEach(f ->
IntStream
.range(1, 4)
.forEach(i -> f.bars.add(new Bar("Bar" + i + " <- " + f.name))));
- 完成上述操作后我们得到三个 foos，每个 foos 包含三个 bars，下面代码将三个对象的 Stream 转化为一个包含九个对象的 Stream
foos.stream()
.flatMap(f -> f.bars.stream())
.forEach(b -> System.out.println(b.name));
// Bar1 <- Foo1
// Bar2 <- Foo1
// Bar3 <- Foo1
// Bar1 <- Foo2
// Bar2 <- Foo2
// Bar3 <- Foo2
// Bar1 <- Foo3
// Bar2 <- Foo3
// Bar3 <- Foo3
- 上述代码可以简化成一个管道流
IntStream.range(1, 4)
.mapToObj(i -> new Foo("Foo" + i))
.peek(f -> IntStream.range(1, 4)
.mapToObj(i -> new Bar("Bar" + i + " <- " f.name))
.forEach(f.bars::add))
.flatMap(f -> f.bars.stream())
.forEach(b -> System.out.println(b.name));
Reduce（减少）
- reduce 操作可以将 Stream 中所有元素组合组合起来得到一个元素，Java8 支持三种不同的 reduce 方法
- 第一种从 Stream 元素序列中提取一个特定的元素，下面代码从用户列表中选择年纪最大的用户操作，下面代码接收一个二元累加器计算函数
persons
.stream()
.reduce((p1, p2) -> p1.age > p2.age ? p1 : p2)
.ifPresent(System.out::println); // Pamela
- 第二种 reduce 接收一个标识值和一个二元操作累加器，这个 reduce 方法把 Stream 中所有用户的名字和年龄汇总到一个新的用户
Person result =
persons
.stream()
.reduce(new Person("", 0), (p1, p2) -> {
p1.age += p2.age;
p1.name += p2.name;
return p1;
});
System.out.format("name=%s; age=%s", result.name, result.age);
// name=MaxPeterPamelaDavid; age=76
- 第三种 reduce 接收三个参数：一个标识值，一个二元操作累加器，一个二元组合方法
Integer ageSum = persons
.stream()
.reduce(0, (sum, p) -> sum += p.age, (sum1, sum2) -> sum1 + sum2);
System.out.println(ageSum); // 76
Parallel Streams（并行流）
- 为了提高大量输入时的执行效率，Stream 可以采用并行或放行执行，通过 ForkJoinPool.commonPool() 方法获取一个可用的 ForkJoinPool，这个 ForkJoinPool 使用5个线程
ForkJoinPool commonPool = ForkJoinPool.commonPool();
System.out.println(commonPool.getParallelism()); // 3
- 可通过 jvm 配置修改
-Djava.util.concurrent.ForkJoinPool.common.parallelism=5
- Collections 中包含 parallelStream() 方法，通过这个方法能够为 Collections 中的元素创建并行流，另外也可以调用 stream 的 parallel() 方法将一个顺序流转换为并行流的拷贝
- 并行流的执行动作，输出中我们可以看到 parallel stream 使用 ForkJoinPool 提供的可用线程来执行并行流
Arrays.asList("a1", "a2", "b1", "c2", "c1")
.parallelStream()
.filter(s -> {
System.out.format("filter: %s [%s]\n",
s, Thread.currentThread().getName());
return true;
})
.map(s -> {
System.out.format("map: %s [%s]\n",
s, Thread.currentThread().getName());
return s.toUpperCase();
})
.forEach(s -> System.out.format("forEach: %s [%s]\n",
s, Thread.currentThread().getName()));
//filter: b1 [main]
//filter: a2 [ForkJoinPool.commonPool-worker-1]
//map: a2 [ForkJoinPool.commonPool-worker-1]
//filter: c2 [ForkJoinPool.commonPool-worker-3]
//map: c2 [ForkJoinPool.commonPool-worker-3]
//filter: c1 [ForkJoinPool.commonPool-worker-2]
//map: c1 [ForkJoinPool.commonPool-worker-2]
//forEach: C2 [ForkJoinPool.commonPool-worker-3]
//forEach: A2 [ForkJoinPool.commonPool-worker-1]
//map: b1 [main]
//forEach: B1 [main]
//filter: a1 [ForkJoinPool.commonPool-worker-3]
//map: a1 [ForkJoinPool.commonPool-worker-3]
//forEach: A1 [ForkJoinPool.commonPool-worker-3]
//forEach: C1 [ForkJoinPool.commonPool-worker-2]
- sort 操作只能在主线程中顺序执行，可以使用 Arrays.parallelSort() 方法来并行执行
- parallel streams 可以带来较大的性能提升，但对于 reduce、collect（组合操作） 需要额外的计算，这时并不如顺序流好用
Arrays.asList("a1", "a2", "b1", "c2", "c1")
.parallelStream()
.filter(s -> {
System.out.format("filter: %s [%s]\n",
s, Thread.currentThread().getName());
return true;
})
.map(s -> {
System.out.format("map: %s [%s]\n",
s, Thread.currentThread().getName());
return s.toUpperCase();
})
.sorted((s1, s2) -> {
System.out.format("sort: %s <> %s [%s]\n",
s1, s2, Thread.currentThread().getName());
return s1.compareTo(s2);
})
.forEach(s -> System.out.format("forEach: %s [%s]\n",
s, Thread.currentThread().getName()));
//filter: c2 [ForkJoinPool.commonPool-worker-3]
//filter: c1 [ForkJoinPool.commonPool-worker-2]
//map: c1 [ForkJoinPool.commonPool-worker-2]
//filter: a2 [ForkJoinPool.commonPool-worker-1]
//map: a2 [ForkJoinPool.commonPool-worker-1]
//filter: b1 [main]
//map: b1 [main]
//filter: a1 [ForkJoinPool.commonPool-worker-2]
//map: a1 [ForkJoinPool.commonPool-worker-2]
//map: c2 [ForkJoinPool.commonPool-worker-3]
//sort: A2 <> A1 [main]
//sort: B1 <> A2 [main]
//sort: C2 <> B1 [main]
//sort: C1 <> C2 [main]
//sort: C1 <> B1 [main]
//sort: C1 <> C2 [main]
//forEach: A1 [ForkJoinPool.commonPool-worker-1]
//forEach: C2 [ForkJoinPool.commonPool-worker-3]
//forEach: B1 [main]
//forEach: A2 [ForkJoinPool.commonPool-worker-2]
//forEach: C1 [ForkJoinPool.commonPool-worker-1]
Stream 常用方法概述：
- stream：为集合创建顺序流
- parallelStream：为集合创建并行流
- forEach：迭代流中的每个数据
- map：映射每个元素到对应结果
- collect：将元素映射为另一种形式
---- 待完善 ----
