[关闭]
@changedi 2017-02-04T08:30:15.000000Z 字数 26007 阅读 5233

一起爪哇Java 8(三)——好用的Stream

Java



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. action.accept(s.get());
  5. return true;
  6. }

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

使用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. @Override
  24. public boolean hasNext() {
  25. return true;
  26. }
  27. @Override
  28. public T next() {
  29. return t = (t == Streams.NONE) ? seed : f.apply(t);
  30. }
  31. };
  32. return StreamSupport.stream(Spliterators.spliteratorUnknownSize(
  33. iterator,
  34. Spliterator.ORDERED | Spliterator.IMMUTABLE), false);
  35. }

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

流的处理

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

中间操作

方法 参数 用途
concat Stream< ? extends T> a, Stream< ? extends T> b 将两个流合起来形成新流
distinct 将流里的元素按照Ojbect.equal方法进行聚合去重,返回一个去重结果的新流
empty 返回一个空的流
filter Predicate< ? super T> predicate 按照谓词参数过滤,过滤后的元素形成新流返回
flatMap Function< ? super T, ? extends Stream< ? extends R>> mapper 将流里的元素T,按照参数Function进行处理,处理结果是一个子流Stream< ? extends R>,后续将子流flat打平,形成元素R的新流。类似的有flatToDouble、flatToInt和flatToLong
limit long maxSize 返回一个新流,只包含maxSize个元素,其他被truncate掉
map Function< ? super T, ? extends R> mapper 经典的map操作,对流里的每个元素,通过参数mapper映射为一个新的元素,返回新元素的流。类似map有mapToDouble、mapToInt和mapToLong
peek Consumer< ? super T> action 这个动作非常有趣,它并不改变流,而是对流里的每个元素执行一个Consumer,对其进行一次处理。原始流不变继续返回
skip long n 跳过n个元素,从第n+1个元素开始返回一个新的流
sorted Comparator< ? super T> comparator 根据参数排序器对流进行排序,返回新的流。如果参数为空,则按照自然序排序

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

终止操作

方法 参数 用途
allMatch Predicate< ? super T> predicate 根据谓词函数判断流里的元素是否都满足,返回对应的boolean值
anyMatch Predicate< ? super T> predicate 根据谓词函数判断流里的元素是否存在一个或多个满足,返回对应的boolean值
noneMatch Predicate< ? super T> predicate 根据谓词函数判断流里的元素是否不存在任何一个元素满足,返回对应的boolean值
count 返回这个流里元素的个数
findAny 返回一个Optional对象,这个等价于对于一个流执行一个select操作,返回一条记录
findFirst 返回这个流里的第一个元素的Optional,如果这个流不是有序的,则返回任意元素
forEach Consumer< ? super T> action 对这个流的每个元素,执行参数Consumer
forEachOrdered Consumer action 针对forEach在并行流里对有序元素的输出不足,这个方法确保并行流中按照原来顺序处理
max Comparator comparator 返回一个Optional值,包含了流里元素的max,条件是按照参数排序器排序
min Comparator comparator 返回一个Optional值,包含了流里元素的min,条件是按照参数排序器排序
reduce BinaryOperator< T> accumulator 经典的reduce,就是根据一个二元操作算子,将流中的元素逐个累计操作一遍,初始元素以foundAny结果为主
reduce T identity, BinaryOperator< T> accumulator 与上面的方法一致,只不过多了一个初始值,不需要findAny了
reduce U identity,BiFunction< U, ? super T, U> accumulator,BinaryOperator< U> combiner 最复杂的reduce,看到combiner会不会有联想?它做的也是对于一个流里的元素T,使用二元函数accumulator计算,计算的值累计到U上,因为之前的reduce要求流元素和结果元素类型一致,所以有限制。而该reduce函数,支持T和U类型不同,通过二元函数转换,但是要求combiner会执行这个事情,要求“ combiner.apply(u, accumulator.apply(identity, t)) == accumulator.apply(u, t)”
collect Supplier< R> supplier,BiConsumer< R, ? super T> accumulator,BiConsumer< R, R> combiner 超级强大的方法。常规的reduce是返回不可变的值。而collect可以将reduce后的值升级为一个可变容器。具体这个方法就是对流里每个元素T,将Supplier提供的值R作为初始值,用BiConsumer的accumulator进行累加计算。combiner的作用和要求和reduce是一样的
collect Collector< ? super T, A, R> collector 和上面的collect一致,只不过Collector封装了一组上面的参数,T是流里的元素,A是累计中间结果,R是返回值的类型(collect的话就是容器了)

这里我们需要看几个方法例子:
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. //增强reduce的
  10. String[] wordArr = {"a", "b", "c"};
  11. Stream<String> words = Stream.of(wordArr);//string 流
  12. //计算每个元素的长度之和,0是初始值,第一次运算后,将其与第一个元素的长度1加和,得到1,把1和原始0进行combine,再不断迭代。。。
  13. int result = words.reduce(0, (length, t) -> t.length() + length, (length1, length2) -> length1 + length2);
  14. 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. if (downstream.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)) {
  60. return new CollectorImpl<>(mangledFactory, accumulator, merger, CH_ID);
  61. }
  62. else {
  63. @SuppressWarnings("unchecked")
  64. Function<A, A> downstreamFinisher = (Function<A, A>) downstream.finisher();
  65. Function<Map<K, A>, M> finisher = intermediate -> {
  66. intermediate.replaceAll((k, v) -> downstreamFinisher.apply(v));
  67. @SuppressWarnings("unchecked")
  68. M castResult = (M) intermediate;
  69. return castResult;
  70. };
  71. return new CollectorImpl<>(mangledFactory, accumulator, merger, finisher, CH_NOID);
  72. }
  73. }

主要对比一下泛型参数吧,这里的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. // Accumulate names into a TreeSet
  4. Set<String> set = people.stream().map(Person::getName).collect(Collectors.toCollection(TreeSet::new));
  5. // Convert elements to strings and concatenate them, separated by commas
  6. String joined = things.stream()
  7. .map(Object::toString)
  8. .collect(Collectors.joining(", "));
  9. // Compute sum of salaries of employee
  10. int total = employees.stream()
  11. .collect(Collectors.summingInt(Employee::getSalary)));
  12. // Group employees by department
  13. Map<Department, List<Employee>> byDept
  14. = employees.stream()
  15. .collect(Collectors.groupingBy(Employee::getDepartment));
  16. // Compute sum of salaries by department
  17. Map<Department, Integer> totalByDept
  18. = employees.stream()
  19. .collect(Collectors.groupingBy(Employee::getDepartment,
  20. Collectors.summingInt(Employee::getSalary)));
  21. // Partition students into passing and failing
  22. Map<Boolean, List<Student>> passingFailing =
  23. students.stream()
  24. .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. @Override
  13. public void accept(P_OUT u) {
  14. if (predicate.test(u))
  15. downstream.accept(u);
  16. }
  17. };
  18. }
  19. };
  20. }

同理分析,它返回的也是个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. // If the input is already naturally sorted and this operation
  4. // also naturally sorted then this is a no-op
  5. if (StreamOpFlag.SORTED.isKnown(flags) && isNaturalSort)
  6. return sink;
  7. else if (StreamOpFlag.SIZED.isKnown(flags))
  8. return new SizedRefSortingSink<>(sink, comparator);
  9. else
  10. return new RefSortingSink<>(sink, comparator);
  11. }

也就是说,如果有序,直接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. RefSortingSink(Sink<? super T> sink, Comparator<? super T> comparator) {
  4. super(sink, comparator);
  5. }
  6. @Override
  7. public void begin(long size) {
  8. if (size >= Nodes.MAX_ARRAY_SIZE)
  9. throw new IllegalArgumentException(Nodes.BAD_SIZE);
  10. list = (size >= 0) ? new ArrayList<T>((int) size) : new ArrayList<T>();
  11. }
  12. @Override
  13. public void end() {
  14. list.sort(comparator);
  15. downstream.begin(list.size());
  16. if (!cancellationWasRequested) {
  17. list.forEach(downstream::accept);
  18. }
  19. else {
  20. for (T t : list) {
  21. if (downstream.cancellationRequested()) break;
  22. downstream.accept(t);
  23. }
  24. }
  25. downstream.end();
  26. list = null;
  27. }
  28. @Override
  29. public void accept(T t) {
  30. list.add(t);
  31. }
  32. }

最终构造的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. return isParallel()
  14. ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
  15. : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
  16. }

其入参是一个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. for ( @SuppressWarnings("rawtypes") AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) {
  4. sink = p.opWrapSink(p.previousStage.combinedFlags, sink);
  5. }
  6. return (Sink<P_IN>) sink;
  7. }

构造好后,执行copyInto:

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

具体核心就是驱动源头的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模式抽象操作和函数,这对于一些以操作为主的系统架构和设计来说,引入方法抽象和函数将极大地辅助系统设计。

添加新批注
在作者公开此批注前,只有你和作者可见。
回复批注