转载自：https://www.zybuluo.com/changedi/note/622375

• 一起爪哇Java 8（三）——好用的Stream
  • Stream组成
    • 源头（Source）
      • Arrays
        • Spliterator
      • Collection
      • 生成方法
    • 流的处理
      • 中间操作
      • 终止操作
  • Stream究竟怎么实现的
    • 一个例子
      • 分析map阶段
      • 分析filter阶段
      • 分析排序阶段
      • 分析limit阶段
      • 分析reduce阶段
  • 总结

Stream组成

在传统Java编程，或者说是类C语言编程中，我们如何操作一个数组数据呢？或者更泛化的讲，我们如何操作一个“集合”（Collection）数据呢？在Java中我们利用java.util包里的各种数据结构封装，来很好的表示了数组（Array）、集合（Set）、列表（List）和kv对象（Map）。但是抽象表示只是意味着存储和定义操作方法，具体如何访问中间的数据，其实还是比较原始的，或者换句话说，操作一个Collection的数据，我们使用的是Collection本身提供的API。就如我们访问一个List里的所有数据，需要一个for循环来get每个element。

Java 8引入了一个Stream对象，来重新封装集合中的数据，就像集合根据其特定的数据结构存储了数据，而Stream将其表示为一个数据流，一个类似List的有序的数据流。值得一提的是，Stream是不存储数据的，它核心是要将不同的数据——流化。

Stream包含一个数据源头（Source），一组（0个或多个）中间操作和一个终止操作。其实很好理解，一个流一定需要一个数据源头，毕竟要确定是哪些数据要流式处理。中间操作是一些类似map、filter之类的转换操作，也就是说map和filter只是将一个流变为新的流，它们可以串起来（stream pipeline）。而终止操作顾名思义，终止操作会结束流，终止操作包括产出结果型和边际效果型（side-effect），其中前者比如count之类的产出一个int值的，后者则是forEach之类的允许后续处理的。下面具体分开讲解一下Stream的各个组成部分。

源头（Source）

其中源头来源于数组、Collection、I/O资源和生成函数。

Arrays

通过一个数组生成一个流，是比较容易理解的。Java API也是通过Arrays.stream()方法来实现的：

1. public static <T> Stream<T> stream(T[] array, int startInclusive, int endExclusive) {
2. return StreamSupport.stream(spliterator(array, startInclusive, endExclusive), false);
3. }

看其声明就是将一个数组转换为一个Stream对象。其委托StreamSupport来构造，而StreamSupport的stream方法声明如下：

1. public static <T> Stream<T> stream(Spliterator<T> spliterator, boolean parallel) {
2. Objects.requireNonNull(spliterator);
3. return new ReferencePipeline.Head<>(spliterator,
4. StreamOpFlag.fromCharacteristics(spliterator),
5. parallel);
6. }

这里一个很重要的参数是Spliterator，这是构造流的核心。ReferencePipeline.Head构造方法就是将一些属性设置好。而ReferencePipeline是个重要的概念，它是pipeline里对于中间操作和源头实现的抽象类，从其类声明中可以看出：

1. abstract class ReferencePipeline<P_IN, P_OUT>
2. extends AbstractPipeline<P_IN, P_OUT, Stream<P_OUT>>
3. implements Stream<P_OUT> {
4. }

两个泛型表示了pipeline的输入和输出。中间操作的实现很多也是由其实现的，比如map和filter。回到刚才的分析，因为ReferencePipeline构造后没有后续的方法链调用了。所以要理解的就是Spliterator这个东西是什么。

Spliterator

Spliterator是java.util包下的一个接口，抽象的能力就是对于一个源头的遍历（traverse）和分区（partition）的能力。也就是说，通过Spliterator来遍历数据流源头的每个元素（或者一个bulk的批量），也通过它来分区数据将其parallel并行化。看它的名字嘛，split+iterator，就是这个意思。

Spliterator声明了几个方法，其中tryAdvance()方法是单个遍历的能力抽象，forEachRemaining()方法是批量遍历的抽象，而trySplit是分区的抽象。这时回去看刚才StreamSupport.stream的参数，其中的Spliterator方法在Arrays类的声明如下：

1. public static <T> Spliterator<T> spliterator(T[] array, int startInclusive, int endExclusive) {
2. return Spliterators.spliterator(array, startInclusive, endExclusive,
3. Spliterator.ORDERED | Spliterator.IMMUTABLE);
4. }

其调用Spliterators静态方法来构造Spliterator，里面的一些ORDERED和IMMUTABLE的常量声明，就是Spliterator的另一部分能力表达——流结构或者源头数据的特征，是有序的还是不可变的？是DISTINCT的还是不可空的？非常多的特性组合。具体到Arrays的调用依赖，Spliterators静态类的实现如下：

1. public static <T> Spliterator<T> spliterator(Object[] array, int fromIndex, int toIndex,
2. int additionalCharacteristics) {
3. checkFromToBounds(Objects.requireNonNull(array).length, fromIndex, toIndex);
4. return new ArraySpliterator<>(array, fromIndex, toIndex, additionalCharacteristics);
5. }

可以看到，返回了一个ArraySpliterator，它实现了针对数组类型数据源头的几个刚才提到的相关方法（tryAdvace和trySplit等）。具体我就不贴代码了，到这个层面大家可以自行关注。

因为我们理解一个流，或者理解一个构造流的过程，其实需要的就是一个Spliterator（我个人很佩服设计者的这种抽象能力），像个schema。

Collection

通过Collection生成流，就能像数组那样静态了，因为Collection是一个接口，标识了一大堆数据结构（Map、List、Set等）。Collection做的比较好的是，流的构建在接口层面完成设计，没有沉到实际的类结构中（全靠Java 8的接口default能力啊）。我们具体看下Collection的构造。

1. default Stream<E> stream() {
2. return StreamSupport.stream(spliterator(), false);
3. }

其同样委托了StreamSupport。通过上面Arrays的分析，我们已经知道了流的构造核心就是Spliterator（第二遍讲了）。所以直接看Spliterator()方法。

1. default Spliterator<E> spliterator() {
2. return Spliterators.spliterator(this, 0);
3. }

也是利用Spliterators的静态方法，只不过传递了Collection自己做参数。

1. public static <T> Spliterator<T> spliterator(Collection<? extends T> c, int characteristics) {
2. return new IteratorSpliterator<>(Objects.requireNonNull(c),
3. characteristics);
4. }

这里就和Arrays不一样了，返回了一个IteratorSpliterator。而具体区别就是Spliterator实现的几个try*方法的不同了。

生成方法

我是这么理解生成方法的，通过generator或iterator来生成。比如generator，主要是通过Stream默认的generate()方法。声明如下：

1. /**
2. * Returns an infinite sequential unordered stream where each element is
3. * generated by the provided {@code Supplier}. This is suitable for
4. * generating constant streams, streams of random elements, etc.
5. *
6. * @param <T> the type of stream elements
7. * @param s the {@code Supplier} of generated elements
8. * @return a new infinite sequential unordered {@code Stream}
9. */
10. public static<T> Stream<T> generate(Supplier<T> s) {
11. Objects.requireNonNull(s);
12. return StreamSupport.stream(
13. new StreamSpliterators.InfiniteSupplyingSpliterator.OfRef<>(Long.MAX_VALUE, s), false);
14. }

这就是一个Supplier来生成流。单单看其中的OfRef()构造方法做的事情，其实就是定义了一个InfiniteSupplyingSpliterator，名曰无穷的供给Spliterator。它的核心tryAdvance方法，就是用tryAdvance的参数Consumer，来执行外层参数Supplier。具体实现如下：

1. @Override
2. public boolean tryAdvance(Consumer<? super T> action) {
3. Objects.requireNonNull(action);
4.  
5. action.accept(s.get());
6. return true;
7. }

这下都明白了。当然场景在注释里也提到了，适合常量流和随机变量流，无穷无尽持续生成。

使用iterate方法构建的流，也是一个无穷的流，声明如下：

1. /**
2. * Returns an infinite sequential ordered {@code Stream} produced by iterative
3. * application of a function {@code f} to an initial element {@code seed},
4. * producing a {@code Stream} consisting of {@code seed}, {@code f(seed)},
5. * {@code f(f(seed))}, etc.
6. *
7. * <p>The first element (position {@code 0}) in the {@code Stream} will be
8. * the provided {@code seed}. For {@code n > 0}, the element at position
9. * {@code n}, will be the result of applying the function {@code f} to the
10. * element at position {@code n - 1}.
11. *
12. * @param <T> the type of stream elements
13. * @param seed the initial element
14. * @param f a function to be applied to to the previous element to produce
15. * a new element
16. * @return a new sequential {@code Stream}
17. */
18. public static<T> Stream<T> iterate(final T seed, final UnaryOperator<T> f) {
19. Objects.requireNonNull(f);
20. final Iterator<T> iterator = new Iterator<T>() {
21. @SuppressWarnings("unchecked")
22. T t = (T) Streams.NONE;
23.  
24. @Override
25. public boolean hasNext() {
26. return true;
27. }
28.  
29. @Override
30. public T next() {
31. return t = (t == Streams.NONE) ? seed : f.apply(t);
32. }
33. };
34. return StreamSupport.stream(Spliterators.spliteratorUnknownSize(
35. iterator,
36. Spliterator.ORDERED | Spliterator.IMMUTABLE), false);
37. }

其通过构造一个Iterator，将遍历方法next实现为调用一元操作算子对初始种子值迭代计算来构造流。

流的处理

流的处理包含了中间操作和终止操作。后面通过两个表格来解释一下中间操作和终止操作的具体方法。

中间操作

中间操作因为不中断流，所以比较好理解，最复杂的算是map，但是也是一个映射关系，因此这里不做例子展示。

终止操作

这里我们需要看几个方法例子： 
reduce:(word length sum)

1. Integer[] digits = {3, 1, 4};
2. Stream<Integer> values = Stream.of(digits);
3. //无初始值的
4. // Optional<Integer> sum = values.reduce((x, y) -> x + y);
5. // System.out.println(sum);// 会输出8
6. //有初始值的
7. Integer sum2 = values.reduce(0, (x, y) -> x + y);
8. System.out.println(sum2);//也会输出8
9.  
10. //增强reduce的
11. String[] wordArr = {"a", "b", "c"};
12. Stream<String> words = Stream.of(wordArr);//string 流
13. //计算每个元素的长度之和，0是初始值，第一次运算后，将其与第一个元素的长度1加和，得到1，把1和原始0进行combine，再不断迭代。。。
14. int result = words.reduce(0, (length, t) -> t.length() + length, (length1, length2) -> length1 + length2);
15. System.out.println("reduce" + result);//这里输出reduce3

原始collect：(word count)

1. String[] wordArr = {"a", "b", "c", "a", "a", "b", "c", "d", "e"};
2. Arrays.stream(wordArr)
3. .collect(TreeMap::new, (map, str) -> {
4. Object val = map.get(str);
5. if (val != null) {
6. Integer v = Integer.valueOf(val.toString());
7. map.put(str, v + 1);
8. } else {
9. map.put(str, 1);
10. }
11. }, (map1, map2) -> {
12. map1.putAll(map2);
13. }).entrySet()
14. .forEach(System.out::println);

输出结果是：

a=3 
b=2 
c=2 
d=1 
e=1

可以看到这样去完成任务还是有点恶心的。这也是Collector想帮我们的地方：

1. String[] wordArr = {"a", "b", "c", "a", "a", "b", "c", "d", "e"};
2. Arrays.stream(wordArr)
3. .collect(groupingBy(Function.identity(), () -> new TreeMap<>(), counting())).entrySet()
4. .forEach(System.out::println);

这是collector版的wordcount，简洁了好多。这里的Collectors.groupingBy是个很复杂的函数，其实现如下：

1. /**
2. * Returns a {@code Collector} implementing a cascaded "group by" operation
3. * on input elements of type {@code T}, grouping elements according to a
4. * classification function, and then performing a reduction operation on
5. * the values associated with a given key using the specified downstream
6. * {@code Collector}. The {@code Map} produced by the Collector is created
7. * with the supplied factory function.
8. *
9. * <p>The classification function maps elements to some key type {@code K}.
10. * The downstream collector operates on elements of type {@code T} and
11. * produces a result of type {@code D}. The resulting collector produces a
12. * {@code Map<K, D>}.
13. *
14. * <p>For example, to compute the set of last names of people in each city,
15. * where the city names are sorted:
16. * <pre>{@code
17. * Map<City, Set<String>> namesByCity
18. * = people.stream().collect(groupingBy(Person::getCity, TreeMap::new,
19. * mapping(Person::getLastName, toSet())));
20. * }</pre>
21. *
22. * @implNote
23. * The returned {@code Collector} is not concurrent. For parallel stream
24. * pipelines, the {@code combiner} function operates by merging the keys
25. * from one map into another, which can be an expensive operation. If
26. * preservation of the order in which elements are presented to the downstream
27. * collector is not required, using {@link #groupingByConcurrent(Function, Supplier, Collector)}
28. * may offer better parallel performance.
29. *
30. * @param <T> the type of the input elements
31. * @param <K> the type of the keys
32. * @param <A> the intermediate accumulation type of the downstream collector
33. * @param <D> the result type of the downstream reduction
34. * @param <M> the type of the resulting {@code Map}
35. * @param classifier a classifier function mapping input elements to keys
36. * @param downstream a {@code Collector} implementing the downstream reduction
37. * @param mapFactory a function which, when called, produces a new empty
38. * {@code Map} of the desired type
39. * @return a {@code Collector} implementing the cascaded group-by operation
40. *
41. * @see #groupingBy(Function, Collector)
42. * @see #groupingBy(Function)
43. * @see #groupingByConcurrent(Function, Supplier, Collector)
44. */
45. public static <T, K, D, A, M extends Map<K, D>>
46. Collector<T, ?, M> groupingBy(Function<? super T, ? extends K> classifier,
47. Supplier<M> mapFactory,
48. Collector<? super T, A, D> downstream) {
49. Supplier<A> downstreamSupplier = downstream.supplier();
50. BiConsumer<A, ? super T> downstreamAccumulator = downstream.accumulator();
51. BiConsumer<Map<K, A>, T> accumulator = (m, t) -> {
52. K key = Objects.requireNonNull(classifier.apply(t), "element cannot be mapped to a null key");
53. A container = m.computeIfAbsent(key, k -> downstreamSupplier.get());
54. downstreamAccumulator.accept(container, t);
55. };
56. BinaryOperator<Map<K, A>> merger = Collectors.<K, A, Map<K, A>>mapMerger(downstream.combiner());
57. @SuppressWarnings("unchecked")
58. Supplier<Map<K, A>> mangledFactory = (Supplier<Map<K, A>>) mapFactory;
59.  
60. if (downstream.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)) {
61. return new CollectorImpl<>(mangledFactory, accumulator, merger, CH_ID);
62. }
63. else {
64. @SuppressWarnings("unchecked")
65. Function<A, A> downstreamFinisher = (Function<A, A>) downstream.finisher();
66. Function<Map<K, A>, M> finisher = intermediate -> {
67. intermediate.replaceAll((k, v) -> downstreamFinisher.apply(v));
68. @SuppressWarnings("unchecked")
69. M castResult = (M) intermediate;
70. return castResult;
71. };
72. return new CollectorImpl<>(mangledFactory, accumulator, merger, finisher, CH_NOID);
73. }
74. }

主要对比一下泛型参数吧，这里的T就是stream元素String，K呢也是String，M是TreeMap，D是counting的结果Long，A也是counting的Supplier等于Long。

by the way，Collectors类还提供了很多静态方法方便开发者将stream做collect操作，比如快捷的toSet/toMap/toList，groupingBy/counting/joining等等。

官方jdk注释里提供了一些简单例子，我这里copy一下：

1. // Accumulate names into a List
2. List<String> list = people.stream().map(Person::getName).collect(Collectors.toList());
3.  
4. // Accumulate names into a TreeSet
5. Set<String> set = people.stream().map(Person::getName).collect(Collectors.toCollection(TreeSet::new));
6.  
7. // Convert elements to strings and concatenate them, separated by commas
8. String joined = things.stream()
9. .map(Object::toString)
10. .collect(Collectors.joining(", "));
11.  
12. // Compute sum of salaries of employee
13. int total = employees.stream()
14. .collect(Collectors.summingInt(Employee::getSalary)));
15.  
16. // Group employees by department
17. Map<Department, List<Employee>> byDept
18. = employees.stream()
19. .collect(Collectors.groupingBy(Employee::getDepartment));
20.  
21. // Compute sum of salaries by department
22. Map<Department, Integer> totalByDept
23. = employees.stream()
24. .collect(Collectors.groupingBy(Employee::getDepartment,
25. Collectors.summingInt(Employee::getSalary)));
26.  
27. // Partition students into passing and failing
28. Map<Boolean, List<Student>> passingFailing =
29. students.stream()
30. .collect(Collectors.partitioningBy(s -> s.getGrade() >= PASS_THRESHOLD));

Stream究竟怎么实现的

一个例子

我们拿一个简单的例子来详细分析一下：

1. String[] a = {"1.0", "2.0", "3.0", "4.0", "5.0"};
2. Optional optional = Stream.of(a).map((v) -> Double.valueOf(v)).filter((v) -> v > 2).sorted((v1, v2) -> v2.compareTo(v1)).limit(2).reduce((v1, v2) -> v1 + v2);
3. System.out.println(optional);

这个例子会输出一个Optional值，答案是12，流的处理逻辑是： 
1. 将数组a构建为一个流：{“1.0”, “2.0”, “3.0”, “4.0”, “5.0”} 
2. 把string型数据映射为double：{1.0, 2.0, 3.0, 4.0, 5.0} 
3. 过滤其中大于2的元素：{3.0, 4.0, 5.0} 
4. 倒序排列元素：{5.0, 4.0, 3.0} 
5. 限制流大小只取前两个元素：{5.0, 4.0} 
6. reduce求和：5.0+4.0=9.0

分析map阶段

我们之前已经讲过构造流源头的分析，产生了一个ReferencePipeline的Head。那么从后续的map开始分析：

1. public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {
2. Objects.requireNonNull(mapper);
3. return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
4. StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
5. @Override
6. Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
7. return new Sink.ChainedReference<P_OUT, R>(sink) {
8. @Override
9. public void accept(P_OUT u) {
10. downstream.accept(mapper.apply(u));
11. }
12. };
13. }
14. };
15. }

仔细看，map返回的是一个StatelessOp，它继承了ReferencePipeline。我们知道ReferencePipeline继承了AbstractPipeline，而AbstractPipeline是真正流结构的持有者。AbstractPipeline里定义了source、nextStage和previousStage，这个结构很明确的将流等同于一个链表，每个node都和前后有关联，并且每个node都有一个到链表起始head的连接。如下图所示：

nextsourceprevioussourcenextprevioussourceHeadop1op2

AbstractPipeline有个onWrapSink抽象方法，其核心就是将每个pipeline表达为一个Sink，而一个流就是由一个源头Source，多个Sink串起来的一个链表结构。每个Sink结构上作为一个链表节点存在，功能上等价于一个Consumer（Sink接口继承Consumer），通过accept方法来执行Sink操作。额外的多了begin和end方法做一些预处理和后处理的工作。

回到刚才的map方法，map返回一个Sink（具体是一个ChainedReference），其中accept做的工作就是调用mapper的apply方法，相当于执行mapper了。做一个拓扑图，就是

nextsourceprevioussourceHeadSink1-mapper

分析filter阶段

接下来构造filter，源码如下：

1. public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
2. Objects.requireNonNull(predicate);
3. return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
4. StreamOpFlag.NOT_SIZED) {
5. @Override
6. Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
7. return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {
8. @Override
9. public void begin(long size) {
10. downstream.begin(-1);
11. }
12.  
13. @Override
14. public void accept(P_OUT u) {
15. if (predicate.test(u))
16. downstream.accept(u);
17. }
18. };
19. }
20. };
21. }

同理分析，它返回的也是个StatelessOp，其中同样是一个ChainedReference的 Sink，accept方法就是会判断一下谓词函数predicate是否接受参数，如果接受调用下游的accept。拓扑变更为

nextsourceprevioussourcenextprevioussourceHeadSink1-mapperSink2-filter

分析排序阶段

再接下来排序，排序返回的是一个StatefulOp，对比之前的map和filter，排序是一个有状态的中间操作。实际的StatefulOp是一个它的子类OfRef，OfRef持有一个排序器

1. private static final class OfRef<T> extends ReferencePipeline.StatefulOp<T, T> {
2. /**
3. * Comparator used for sorting
4. */
5. private final boolean isNaturalSort;
6. private final Comparator<? super T> comparator;
7. ...
8. }

它的onWrapSink方法返回

1. public Sink<T> opWrapSink(int flags, Sink<T> sink) {
2. Objects.requireNonNull(sink);
3.  
4. // If the input is already naturally sorted and this operation
5. // also naturally sorted then this is a no-op
6. if (StreamOpFlag.SORTED.isKnown(flags) && isNaturalSort)
7. return sink;
8. else if (StreamOpFlag.SIZED.isKnown(flags))
9. return new SizedRefSortingSink<>(sink, comparator);
10. else
11. return new RefSortingSink<>(sink, comparator);
12. }

也就是说，如果有序，直接pass，返回原sink，否则，根据其列表是否有界，来使用不同的Sink返回，有界的排序Sink内部是一个数组T[]，而无界的排序Sink内部是一个ArrayList< T>。以无界排序Sink为例，其中的三个方法begin负责构造List，accept负责把流元素add进去，而end负责排序sort。看下实现一目了然：

1. private static final class RefSortingSink<T> extends AbstractRefSortingSink<T> {
2. private ArrayList<T> list;
3.  
4. RefSortingSink(Sink<? super T> sink, Comparator<? super T> comparator) {
5. super(sink, comparator);
6. }
7.  
8. @Override
9. public void begin(long size) {
10. if (size >= Nodes.MAX_ARRAY_SIZE)
11. throw new IllegalArgumentException(Nodes.BAD_SIZE);
12. list = (size >= 0) ? new ArrayList<T>((int) size) : new ArrayList<T>();
13. }
14.  
15. @Override
16. public void end() {
17. list.sort(comparator);
18. downstream.begin(list.size());
19. if (!cancellationWasRequested) {
20. list.forEach(downstream::accept);
21. }
22. else {
23. for (T t : list) {
24. if (downstream.cancellationRequested()) break;
25. downstream.accept(t);
26. }
27. }
28. downstream.end();
29. list = null;
30. }
31.  
32. @Override
33. public void accept(T t) {
34. list.add(t);
35. }
36. }

最终构造的Stream拓扑结构变为

nextsourceprevioussourcenextprevioussourcenextprevioussourceHeadSink1-mapperSink2-filterSink3-sorted

分析limit阶段

limit也是一个StatefulOp，其和skip复用了一个Sink，其中的onWrapSink方法具体会在accept阶段控制一个limit的size：

1. public void accept(T t) {
2. if (n == 0) {
3. if (m > 0) {
4. m--;
5. downstream.accept(t);
6. }
7. }
8. else {
9. n--;
10. }
11. }

其他的逻辑和之前的sink类似，不赘述，拓扑结构更新如下：

nextsourceprevioussourcenextprevioussourcenextprevioussourcenextprevioussourceHeadSink1-mapperSink2-filterSink3-sortedSink4-limit

分析reduce阶段

reduce就和之前的Sink不一样了，reduce作为一个终止操作，不再是Sink，而是一个触发流执行的动作，其调用了AbstractPipeline的evaluate方法来触发整个流的执行。

1. public final Optional<P_OUT> reduce(BinaryOperator<P_OUT> accumulator) {
2. return evaluate(ReduceOps.makeRef(accumulator));
3. }

evaluate方法声明如下：

1. /**
2. * Evaluate the pipeline with a terminal operation to produce a result.
3. *
4. * @param <R> the type of result
5. * @param terminalOp the terminal operation to be applied to the pipeline.
6. * @return the result
7. */
8. final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {
9. assert getOutputShape() == terminalOp.inputShape();
10. if (linkedOrConsumed)
11. throw new IllegalStateException(MSG_STREAM_LINKED);
12. linkedOrConsumed = true;
13.  
14. return isParallel()
15. ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
16. : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
17. }

其入参是一个TerminalOp终止操作，而返回的就是流执行后的结果。干的事情就是调用TerminalOp的evaluateSequential方法来执行整个流。具体到这个例子就是调用ReduceOp（它继承TerminalOp）的evaluateSequential方法。

1. public <P_IN> R evaluateSequential(PipelineHelper<T> helper,
2. Spliterator<P_IN> spliterator) {
3. return helper.wrapAndCopyInto(makeSink(), spliterator).get();
4. }

其中PipelineHelper的wrapAndCopyInto方法就是包装sink，然后copyInto就是执行了。

1. final <P_IN, S extends Sink<E_OUT>> S wrapAndCopyInto(S sink, Spliterator<P_IN> spliterator) {
2. copyInto(wrapSink(Objects.requireNonNull(sink)), spliterator);
3. return sink;
4. }

其中wrapSink很重要，它负责调用之前拓扑结构的每个Sink的onWrapSink方法把这个链式结构建立起来。看代码会发现，链是自后向前的。

1. final <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) {
2. Objects.requireNonNull(sink);
3.  
4. for ( @SuppressWarnings("rawtypes") AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) {
5. sink = p.opWrapSink(p.previousStage.combinedFlags, sink);
6. }
7. return (Sink<P_IN>) sink;
8. }

构造好后，执行copyInto：

1. final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
2. Objects.requireNonNull(wrappedSink);
3.  
4. if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
5. wrappedSink.begin(spliterator.getExactSizeIfKnown());
6. spliterator.forEachRemaining(wrappedSink);
7. wrappedSink.end();
8. }
9. else {
10. copyIntoWithCancel(wrappedSink, spliterator);
11. }
12. }

具体核心就是驱动源头的Spliterator开始forEachRemaining，也就是遍历。对于每个元素进行tryAdvance(action)，此处的action就是一个Sink了（刚才说过Sink继承Consumer）。这样一个链就开始从头到尾（不一定，这取决于源头结构和Spliterator类型）执行了。

最终的链式拓扑结构如下：

nextsourceprevioussourcenextprevioussourcenextprevioussourcenextprevioussource无真实关联HeadSink1-mapperSink2-filterSink3-sortedSink4-limitTerminalOp-reduce

链式的执行依赖于TerminalOp究竟是什么，因此一个Stream的执行是lazy的，当流构建好时，只是一个Sink的链式结构，最终的遍历和执行需要一个终止操作来触发。回到本节开始时的例子，前5步流是构造不执行的，直到第6步确定reduce求和，才触发了流的遍历执行。

总结

本文是Java8三个系列的最后一篇，笔者非常粗浅地分析了Java 8的Stream用法和简单实现剖析。整个一个系列完成，个人以为Java 8的三大亮点也分别简单总结整理完成。对于Java开发者，个人建议要在程序设计时广泛使用流式Stream用法，其可以非常高效地帮助开发者完成一些常用结构体（array/list/map）的遍历操作；使用函数式设计，可以极大的补充面向对象的编程思维，在OO模式抽象实体的同时，可以FP模式抽象操作和函数，这对于一些以操作为主的系统架构和设计来说，引入方法抽象和函数将极大地辅助系统设计。