Java8 Stream API 介绍
Java 从版本8开始支持“Stream API”，即函数式编程，可以用简单的代码表达出比较复杂的遍历操作。本文介绍这些 Stream API 的基本概念，用法，以及一些参考资料。我之前写 Python 比较多，所以一些地方可能用 Python 的视角来解释。
简单用法
public class streamApi {
public static void main(String[] args) {
List<String> myArray =
Arrays.asList("a1", "a2", "b1", "b2", "c1", "c2");
myArray.stream()
.filter(s -> s.startsWith("b"))
.map(String::toUpperCase)
.sorted()
.forEach(System.out::print);
}
}
这些函数和 Python 的 filter
map
sort
很像了，所以很容看懂。就是先过滤出以 "b"
开头的字符串，然后用 map
转换成大写的方式，排序之后，输出。
Stream API 中有一个概念，将这些 API 分成了两种：
-
中间结果（intermediate）：像
filtermapsorted的结果都是中间结果，可以继续使用 Stream API 连续调用； -
最终结果（terminal）：类似
forEach，这种 API 将会终止 Stream 的调用，它们的结果要么是void，要么是一个非 Stream 结果。
连续调用 Stream 的方式叫做 operation pipeline 。所有的 Stream API 可以参考这个 javadoc 。
大多数的 Stream 操作都接收一个 lambda 表达式作为参数，Lambda 表达式描述了 Stream 操作的一个具体行为，通常都是 stateless 和 non-interfering 的。
Non-interfering
意味着它不会修改原始的数据，比如上面的例子，没有操作去动 myList
，迭代结束之后， myList
还是保持着原来的样子。
Stateless 意味着操作都是确定的，没有依赖外面的变量（导致可能在执行期间改变）。
不同类型的 Stream 操作
Stream 可以从不同的数据类型创建，尤其是集合（Collections）。
List 和 Set 支持 stream()
方法和 parallelStream()
方法， parallenStream()
可以在多线程中执行。
在 List 上调用 stream()
可以创建一个 Stream 对象。
Arrays.asList("a1", "a2", "a3")
.stream()
.findFirst()
.ifPresent(System.out::println); // a1
但是我们不必专门为了创建 Stream 对象而创建一个集合：
Stream.of("a1", "a2", "a3")
.findFirst()
.ifPresent(System.out::println); // a1
Stream.of()
可以从一些对象引用中自动创建一个 Stream。
除了从对象创建 Stream，Java8 还提供了方法从基本类型创建 Stream，比如 int
long
double
，这些方法分别是 IntStream
LongStream
DoubleStream
.
IntStream 可以用来替代 for 循环：
IntStream.range(1, 4) .forEach(System.out::println); // 1 // 2 // 3
基本类型的 Stream 和普通的 Stream 对象基本一样，几点区别如下：
-
基本类型使用特殊的 lambda 表达式，比如
IntFunction之于Function,IntPredicate之于Predicate; -
基本类型支持一些特殊的“最终结果API”，比如
sum()average()；
Arrays.stream(new int[] {1, 2, 3})
.map(n -> 2 * n + 1)
.average()
.ifPresent(System.out::println); // 5.0
有时候，我们想把普通的 Stream 转换成原始类型的 Stream，比如我们想用 max()
，这时可以使用转换的方法 mapToInt()
mapToLong()
mapToDouble()
:
Stream.of("a1", "a2", "a3")
.map(s -> s.substring(1))
.mapToInt(Integer::parseInt)
.max()
.ifPresent(System.out::println); // 3
原始类型可以通过 mapToObject
方法，将原始类型的 Stream 转换成普通的 Stream 对象。
IntStream.range(1, 4) .mapToObj(i -> "a" + i) .forEach(System.out::println); // a1 // a2 // a3
下面这个例子结合了普通的 Stream 和原始类型的 Stream：
Stream.of(1.0, 2.0, 3.0) .mapToInt(Double::intValue) .mapToObj(i -> "a" + i) .forEach(System.out::println); // a1 // a2 // a3
执行顺序
前面介绍了 Stream 的基本概念，下面开始深入原理。
产生中间结果的 Stream 一个比较重要的特性是，它是惰性的。下面这个例子，我们只有中间结果，没有最终结果的 Stream，最终 println
不会被执行。
Stream.of("d2", "a2", "b1", "b3", "c")
.filter(s -> {
System.out.println("filter: " + s);
return true;
});
因为 Stream 操作是惰性的，只有用到的时候才会真正执行。
如果我们在后面加上一个终止类型的 Stream 操作， println
就会执行了。
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
注意从输出的顺序也可以看到“惰性执行”的特征：并不是所有的 filter
都打印出来，再打印出来 forEeach
。而是一个元素执行到底，再去执行下一个元素。
这样可以减少执行的次数。参考下面这个例子：
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
anyMatch
会在找到第一个符合条件的元素就返回。这样我们并不需要对有的元素执行 map
，在第一个 anyMatch
返回 true
之后，执行就结束了。所以前面的中间状态 Stream 操作，会执行尽可能少的次数。
执行的顺序很重要（Stream 的优化）
下面这个例子，我们用了两个生成中间结果的 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
执行了1次。
如果我们在这里稍微改变一下顺序，将 filter
提前执行，可以将 map
的执行次数减少到1次。（有点像 SQL 优化）
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
现在 map
只执行一次了，在操作很大的集合的时候非常有用。
下面我们引入一下 sorted
这个操作：
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));
排序是一个特殊的中间操作，是一个 stateful 的操作。因为需要原地排序。
输出如下：
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
排序操作会在整个集合上执行。所以和之前的“垂直”执行不同，排序操作是水平执行的。注意排序影响的只是后面的 Stream 操作，对于原来的集合，顺序依然是不变的。参考这段代码：
public class streamApi {
public static void main(String[] args) {
List<String> foo = Arrays.asList("d2", "a2", "b1", "b3", "c");
foo.stream().sorted(
(s1, s2) -> {
return s1.compareTo(s2);
}
).forEach(
System.out::println
);
System.out.println("---");
foo.stream().forEach(System.out::println);
}
}
输出如下（注意原来的 foo
并没有变化）：
a2 b1 b3 c d2 --- d2 a2 b1 b3 c
Sort 也有惰性执行的特性，如果我们改变一下上面那个例子的执行顺序：
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
可以发现 sorted
不会执行，因为 filter
只产生了一个元素。
Stream 的重用
Java8 的 Stream 是不支持重用的。一旦调用了终止类型的 Stream 操作，Stream 会被 close。
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
在同一个 Stream 上，先调用 noneMatch
再调用 anyMatch
会看到以下异常：
java.lang.IllegalStateException: stream has already been operated upon or closed at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:229) at java.util.stream.ReferencePipeline.noneMatch(ReferencePipeline.java:459) at com.winterbe.java8.Streams5.test7(Streams5.java:38) at com.winterbe.java8.Streams5.main(Streams5.java:28)
所以，我们必须为每一个终止类型的 Stream 操作创建一个新的 Stream。可以用 Stream Supplier 来实现。
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
每次调用 get()
都会得到一个新的 Stream。
高级操作
Stream 支持的操作很多（不像Python的函数式编程只支持4个）。我们已经见过了最常用的 filter
和 map
。其他的操作读者可以自行阅读 Stream 文档
。这里，我们再试一下几个复杂的操作： collect
flatMap
reduce
.
下面的例子都会使用一个 Person
的 List 来演示。
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
Collect 是很有用的一个终止类型的 Stream 操作，可以将 Stream 转换成集合结果，比如 List
Set
Map
。Collect 接收一个 Collector
作为参数， Collector
需要支持4种操作：
- supplier
- accumulator
- combiner
- finisher
听起来实现很复杂，但是好处是 Java8 已经内置了常用的 Collector
，所以大多数情况下我们不需要自己实现。
下面看一个常用的操作：
List<Person> filtered =
persons
.stream()
.filter(p -> p.name.startsWith("P"))
.collect(Collectors.toList());
System.out.println(filtered); // [Peter, Pamela]
可以看到这个 Stream 操作最后构建了一个 List，如果需要 Set 的话只需要将 toList()
换成 toSet()
。
接下来这个例子，将对象按照属性存放到 Map
中。
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 非常实用，还可以对 Stream 进行聚合，比如计算所有 Person 的平均年龄：
Double averageAge = persons .stream() .collect(Collectors.averagingInt(p -> p.age)); System.out.println(averageAge); // 19.0
如果需要更全面的统计数据，可以试一下 summarizing
Collector，这个内置的 Collector 提供了 count, sum, min, max 等有用的数据。
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 成一个 String：
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.
Join Collector 的参数是一个分隔符，一个可选的前缀和后缀。
将 Stream 元素转换成 map 的时候，需要特别注意：key 必须是唯一的，否则会抛出 IllegalStateException
。但是我们可以传入一个 merge function，来指定重复的元素映射的方式：
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。前面已经提到过，实现一个 Collector，我们需要提供4个东西：supplier,accumulator,combiner,finisher.
下面这个 Collector 将所有的 Person 对象转换成一个字符串，名字全部大写，中间用 |
分割。
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
Java 的 String 是不可修改的，所以这里需要一个 helper class StringJoiner
，来构建最终的 String。
|
FlatMap
前面我们演示了如何用 map
将一种类型的对象转换成另一种类型。但是 map
也有一些限制：一个对象只能转换成一个对象，如果需要将一个对象转换成多个就不行了。所以还有一个 flatMap
。
FlatMap 可以将 Stream 中的每一个对象转换成0个，1个或多个。无论产生多少对象，最终都会放到同一个 Stream 中，供后面的操作消费。
下面演示 flatMap
的功能，我们需要一个有继承关系的类型：
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;
}
}
下面，我们使用 Stream 来初始化多个几个对象：
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))));
现在我们生成了一个 List，包含3个 foo，每个 foo 中包含3个 bar.
FlatMap 接收一个方法，返回一个 Stream，可以包含任意个 Objects. 所以我们可以用这个方法得到 foo 中的每一个 bar：
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
上面这个代码将 foo 的 Stream 转换成了包含 9 个 bar 的 Stream。
上面所有的代码也可以简化到一个 Stream 操作中：
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));
在 flatMap
中也可以用 Optional
对象，Optional 是 Java8 引入的，可以检查 null 的一种机制。结合 Optional 和 flatMap
我们可以相对优雅地处理 null
，考虑下面这种数据结构：
class Outer {
Nested nested;
}
class Nested {
Inner inner;
}
class Inner {
String foo;
}
为了正确地得到 Inner
中的 foo
String，我们要这么写：
Outer outer = new Outer();
if (outer != null && outer.nested != null && outer.nested.inner != null) {
System.out.println(outer.nested.inner.foo);
}
用 flatMap
的话，我们可以这么写：
Optional.of(new Outer()) .flatMap(o -> Optional.ofNullable(o.nested)) .flatMap(n -> Optional.ofNullable(n.inner)) .flatMap(i -> Optional.ofNullable(i.foo)) .ifPresent(System.out::println);
每一个 flatMap
都用 Optional
封装，如果不是空，就返回里面的对象，如果是空的话就返回一个 null
.
Reduce
Reduce 操作可以将所有的元素编程一个结果。Java8 支持3种不同的 reduce
方法。
第一种可以将 Stream 中的元素聚合成一个。比如下面的代码，可以找到 Stream 中年龄最大的 Person.
persons .stream() .reduce((p1, p2) -> p1.age > p2.age ? p1 : p2) .ifPresent(System.out::println); // Pamela
reduce
方法接收一个二元函数（一个只有两个参数的函数）作为参数，返回一个对象。（所以叫做 reduce）
第二种 reduce
接收一个初始对象，和一个二元函数。通常可以用于聚合操作（比如累加）。
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
方法接收3个参数：一个初始化对象，一个二元函数，和一个 combiner 函数。
初始化值并不一定是 Stream 中的对象，所以我们可以直接用一个整数。
Integer ageSum = persons .stream() .reduce(0, (sum, p) -> sum += p.age, (sum1, sum2) -> sum1 + sum2); System.out.println(ageSum); // 76
结果依然是 76，那么原理是什么呢？我们可以打印出来执行过程：
Integer ageSum = persons
.stream()
.reduce(0,
(sum, p) -> {
System.out.format("accumulator: sum=%s; person=%s/n", sum, p);
return sum += p.age;
},
(sum1, sum2) -> {
System.out.format("combiner: sum1=%s; sum2=%s/n", sum1, sum2);
return sum1 + sum2;
});
// accumulator: sum=0; person=Max
// accumulator: sum=18; person=Peter
// accumulator: sum=41; person=Pamela
// accumulator: sum=64; person=David
可以看到 accumulator 做了所有的工作，将所有的年龄和初始化的 int 值 0 相加。但是 combiner 没有执行？
我们将 Stream 换成 parallelStream 再来看一下：
Integer ageSum = persons
.parallelStream()
.reduce(0,
(sum, p) -> {
System.out.format("accumulator: sum=%s; person=%s/n", sum, p);
return sum += p.age;
},
(sum1, sum2) -> {
System.out.format("combiner: sum1=%s; sum2=%s/n", sum1, sum2);
return sum1 + sum2;
});
// accumulator: sum=0; person=Pamela
// accumulator: sum=0; person=David
// accumulator: sum=0; person=Max
// accumulator: sum=0; person=Peter
// combiner: sum1=18; sum2=23
// combiner: sum1=23; sum2=12
// combiner: sum1=41; sum2=35
这次 combiner 执行了。并发执行的 Stream 有不同的行为。Accumulator 是并发执行的，所以需要一个 combiner 将所有的并发得到的结果再聚合起来。
下面来看一下 Parallel Stream。
Parallel Stream
因为 Stream 中每一个元素都是单独执行的，可想而知，如果并行计算每一个元素的话，可以提升性能。Parallel Stream 就是适用这种场景的。Parallel Stream 使用公共的 ForkJoinPool
来并行计算。底层的真正的线程数据取决于 CPU 的核数，默认是3.
ForkJoinPool commonPool = ForkJoinPool.commonPool(); System.out.println(commonPool.getParallelism()); // 3
这个值可以通过 JVM 参数修改：
-Djava.util.concurrent.ForkJoinPool.common.parallelism=5
Collections 可以通过 parallelStream()
来创建一个并行执行的 Stream，可以在普通的 Stream 上执行 parallel()
来转换成并行执行的 Stream。
下面这个例子，将并行执行的每一步的线程执行者打印出来：
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]
从上面的结果页可以看出，所有的 ForkJoinPool
中的线程都参与了计算。
如果在上面的例子中加入一个 sort
操作，结果就有些不同了：
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]
看起来 sort
好像是顺序执行的。实际上， sort
使用的是 Java8 的 Arrays.parallelSort()
方法， 文档
里提到，这里的排序是否真正的并行执行取决于数组的长度，如果长的话就会用并行排序，否则就用单线程排序：
If the length of the specified array is less than the minimum granularity, then it is sorted using the appropriate Arrays.sort method.
回到之前的 reduce
方法，我们知道 combiner 只会在并行的时候执行，现在来看一下这个方法到底是做什么的：
List<Person> persons = Arrays.asList(
new Person("Max", 18),
new Person("Peter", 23),
new Person("Pamela", 23),
new Person("David", 12));
persons
.parallelStream()
.reduce(0,
(sum, p) -> {
System.out.format("accumulator: sum=%s; person=%s [%s]/n",
sum, p, Thread.currentThread().getName());
return sum += p.age;
},
(sum1, sum2) -> {
System.out.format("combiner: sum1=%s; sum2=%s [%s]/n",
sum1, sum2, Thread.currentThread().getName());
return sum1 + sum2;
});
可以看到 accumulator 和 combiner 都使用了多线程来运行：
accumulator: sum=0; person=Pamela; [main] accumulator: sum=0; person=Max; [ForkJoinPool.commonPool-worker-3] accumulator: sum=0; person=David; [ForkJoinPool.commonPool-worker-2] accumulator: sum=0; person=Peter; [ForkJoinPool.commonPool-worker-1] combiner: sum1=18; sum2=23; [ForkJoinPool.commonPool-worker-1] combiner: sum1=23; sum2=12; [ForkJoinPool.commonPool-worker-2] combiner: sum1=41; sum2=35; [ForkJoinPool.commonPool-worker-2]
综上，在数据量很大的时候，并行执行的 Stream 可以带来很大的性能提升。但是注意像 reduce
和 collect
这样的操作，需要特殊的 combiner。（因为前一操作产生的类型不同，需要做聚合，所以无法和迁移操作的函数一样，需要另外提供）。
另外要注意的是，Parallel Stream 底层使用的通用的 ForkJoinPool
，所以需要注意不要在并行的 Stream 中出现很慢或阻塞的操作，这样会影响其他并行任务。
以上就是基本的 Stream API 介绍了，强烈建议阅读 Java8 的官方文档。
参考资料：
- package summary
- Document
- Toturial
- Java 8 Stream Tutorial
- Collection Pipeline by Martin Fowler
- Stream.js Javascript 版的 Java Stream API
正文到此结束
热门推荐
相关文章
近期评论
-
666
-
文章和留言都翻到11页了 没有OOM
-
-
朋友，翻页到11页，及以后，会出现OOM，无法访问
-
-
-
-
版本号是多少，你可以下载哪个代码仓库，jdk选1.8 直接跑就行
-
-
Loading...
![[HBLOG]公众号](https://www.liuhaihua.cn/img/qrcode_gzh.jpg)
