Java8 Stream

2018-02-27 11:45:51来源:https://www.jianshu.com/p/3b665e774e52作者:林塬人点击

分享


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:将元素映射为另一种形式

---- 待完善 ----








最新文章

123

最新摄影

闪念基因

微信扫一扫

第七城市微信公众平台