@changedi
2017-02-04T16:30:15.000000Z
字数 26007
阅读 9415
Java
在传统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的各个组成部分。
其中源头来源于数组、Collection、I/O资源和生成函数。
通过一个数组生成一个流,是比较容易理解的。Java API也是通过Arrays.stream()方法来实现的:
public static <T> Stream<T> stream(T[] array, int startInclusive, int endExclusive) {
return StreamSupport.stream(spliterator(array, startInclusive, endExclusive), false);
}
看其声明就是将一个数组转换为一个Stream对象。其委托StreamSupport来构造,而StreamSupport的stream方法声明如下:
public static <T> Stream<T> stream(Spliterator<T> spliterator, boolean parallel) {
Objects.requireNonNull(spliterator);
return new ReferencePipeline.Head<>(spliterator,
StreamOpFlag.fromCharacteristics(spliterator),
parallel);
}
这里一个很重要的参数是Spliterator,这是构造流的核心。ReferencePipeline.Head构造方法就是将一些属性设置好。而ReferencePipeline是个重要的概念,它是pipeline里对于中间操作和源头实现的抽象类,从其类声明中可以看出:
abstract class ReferencePipeline<P_IN, P_OUT>
extends AbstractPipeline<P_IN, P_OUT, Stream<P_OUT>>
implements Stream<P_OUT> {
}
两个泛型表示了pipeline的输入和输出。中间操作的实现很多也是由其实现的,比如map和filter。回到刚才的分析,因为ReferencePipeline构造后没有后续的方法链调用了。所以要理解的就是Spliterator这个东西是什么。
Spliterator是java.util包下的一个接口,抽象的能力就是对于一个源头的遍历(traverse)和分区(partition)的能力。也就是说,通过Spliterator来遍历数据流源头的每个元素(或者一个bulk的批量),也通过它来分区数据将其parallel并行化。看它的名字嘛,split+iterator,就是这个意思。
Spliterator声明了几个方法,其中tryAdvance()方法是单个遍历的能力抽象,forEachRemaining()方法是批量遍历的抽象,而trySplit是分区的抽象。这时回去看刚才StreamSupport.stream的参数,其中的Spliterator方法在Arrays类的声明如下:
public static <T> Spliterator<T> spliterator(T[] array, int startInclusive, int endExclusive) {
return Spliterators.spliterator(array, startInclusive, endExclusive,
Spliterator.ORDERED | Spliterator.IMMUTABLE);
}
其调用Spliterators静态方法来构造Spliterator,里面的一些ORDERED和IMMUTABLE的常量声明,就是Spliterator的另一部分能力表达——流结构或者源头数据的特征,是有序的还是不可变的?是DISTINCT的还是不可空的?非常多的特性组合。具体到Arrays的调用依赖,Spliterators静态类的实现如下:
public static <T> Spliterator<T> spliterator(Object[] array, int fromIndex, int toIndex,
int additionalCharacteristics) {
checkFromToBounds(Objects.requireNonNull(array).length, fromIndex, toIndex);
return new ArraySpliterator<>(array, fromIndex, toIndex, additionalCharacteristics);
}
可以看到,返回了一个ArraySpliterator,它实现了针对数组类型数据源头的几个刚才提到的相关方法(tryAdvace和trySplit等)。具体我就不贴代码了,到这个层面大家可以自行关注。
因为我们理解一个流,或者理解一个构造流的过程,其实需要的就是一个Spliterator(我个人很佩服设计者的这种抽象能力),像个schema。
通过Collection生成流,就能像数组那样静态了,因为Collection是一个接口,标识了一大堆数据结构(Map、List、Set等)。Collection做的比较好的是,流的构建在接口层面完成设计,没有沉到实际的类结构中(全靠Java 8的接口default能力啊)。我们具体看下Collection的构造。
default Stream<E> stream() {
return StreamSupport.stream(spliterator(), false);
}
其同样委托了StreamSupport。通过上面Arrays的分析,我们已经知道了流的构造核心就是Spliterator(第二遍讲了)。所以直接看Spliterator()方法。
default Spliterator<E> spliterator() {
return Spliterators.spliterator(this, 0);
}
也是利用Spliterators的静态方法,只不过传递了Collection自己做参数。
public static <T> Spliterator<T> spliterator(Collection<? extends T> c, int characteristics) {
return new IteratorSpliterator<>(Objects.requireNonNull(c),
characteristics);
}
这里就和Arrays不一样了,返回了一个IteratorSpliterator。而具体区别就是Spliterator实现的几个try*方法的不同了。
我是这么理解生成方法的,通过generator或iterator来生成。比如generator,主要是通过Stream默认的generate()方法。声明如下:
/**
* Returns an infinite sequential unordered stream where each element is
* generated by the provided {@code Supplier}. This is suitable for
* generating constant streams, streams of random elements, etc.
*
* @param <T> the type of stream elements
* @param s the {@code Supplier} of generated elements
* @return a new infinite sequential unordered {@code Stream}
*/
public static<T> Stream<T> generate(Supplier<T> s) {
Objects.requireNonNull(s);
return StreamSupport.stream(
new StreamSpliterators.InfiniteSupplyingSpliterator.OfRef<>(Long.MAX_VALUE, s), false);
}
这就是一个Supplier来生成流。单单看其中的OfRef()构造方法做的事情,其实就是定义了一个InfiniteSupplyingSpliterator,名曰无穷的供给Spliterator。它的核心tryAdvance方法,就是用tryAdvance的参数Consumer,来执行外层参数Supplier。具体实现如下:
@Override
public boolean tryAdvance(Consumer<? super T> action) {
Objects.requireNonNull(action);
action.accept(s.get());
return true;
}
这下都明白了。当然场景在注释里也提到了,适合常量流和随机变量流,无穷无尽持续生成。
使用iterate方法构建的流,也是一个无穷的流,声明如下:
/**
* Returns an infinite sequential ordered {@code Stream} produced by iterative
* application of a function {@code f} to an initial element {@code seed},
* producing a {@code Stream} consisting of {@code seed}, {@code f(seed)},
* {@code f(f(seed))}, etc.
*
* <p>The first element (position {@code 0}) in the {@code Stream} will be
* the provided {@code seed}. For {@code n > 0}, the element at position
* {@code n}, will be the result of applying the function {@code f} to the
* element at position {@code n - 1}.
*
* @param <T> the type of stream elements
* @param seed the initial element
* @param f a function to be applied to to the previous element to produce
* a new element
* @return a new sequential {@code Stream}
*/
public static<T> Stream<T> iterate(final T seed, final UnaryOperator<T> f) {
Objects.requireNonNull(f);
final Iterator<T> iterator = new Iterator<T>() {
@SuppressWarnings("unchecked")
T t = (T) Streams.NONE;
@Override
public boolean hasNext() {
return true;
}
@Override
public T next() {
return t = (t == Streams.NONE) ? seed : f.apply(t);
}
};
return StreamSupport.stream(Spliterators.spliteratorUnknownSize(
iterator,
Spliterator.ORDERED | Spliterator.IMMUTABLE), false);
}
其通过构造一个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)
Integer[] digits = {3, 1, 4};
Stream<Integer> values = Stream.of(digits);
//无初始值的
// Optional<Integer> sum = values.reduce((x, y) -> x + y);
// System.out.println(sum);// 会输出8
//有初始值的
Integer sum2 = values.reduce(0, (x, y) -> x + y);
System.out.println(sum2);//也会输出8
//增强reduce的
String[] wordArr = {"a", "b", "c"};
Stream<String> words = Stream.of(wordArr);//string 流
//计算每个元素的长度之和,0是初始值,第一次运算后,将其与第一个元素的长度1加和,得到1,把1和原始0进行combine,再不断迭代。。。
int result = words.reduce(0, (length, t) -> t.length() + length, (length1, length2) -> length1 + length2);
System.out.println("reduce" + result);//这里输出reduce3
原始collect:(word count)
String[] wordArr = {"a", "b", "c", "a", "a", "b", "c", "d", "e"};
Arrays.stream(wordArr)
.collect(TreeMap::new, (map, str) -> {
Object val = map.get(str);
if (val != null) {
Integer v = Integer.valueOf(val.toString());
map.put(str, v + 1);
} else {
map.put(str, 1);
}
}, (map1, map2) -> {
map1.putAll(map2);
}).entrySet()
.forEach(System.out::println);
输出结果是:
a=3
b=2
c=2
d=1
e=1
可以看到这样去完成任务还是有点恶心的。这也是Collector想帮我们的地方:
String[] wordArr = {"a", "b", "c", "a", "a", "b", "c", "d", "e"};
Arrays.stream(wordArr)
.collect(groupingBy(Function.identity(), () -> new TreeMap<>(), counting())).entrySet()
.forEach(System.out::println);
这是collector版的wordcount,简洁了好多。这里的Collectors.groupingBy是个很复杂的函数,其实现如下:
/**
* Returns a {@code Collector} implementing a cascaded "group by" operation
* on input elements of type {@code T}, grouping elements according to a
* classification function, and then performing a reduction operation on
* the values associated with a given key using the specified downstream
* {@code Collector}. The {@code Map} produced by the Collector is created
* with the supplied factory function.
*
* <p>The classification function maps elements to some key type {@code K}.
* The downstream collector operates on elements of type {@code T} and
* produces a result of type {@code D}. The resulting collector produces a
* {@code Map<K, D>}.
*
* <p>For example, to compute the set of last names of people in each city,
* where the city names are sorted:
* <pre>{@code
* Map<City, Set<String>> namesByCity
* = people.stream().collect(groupingBy(Person::getCity, TreeMap::new,
* mapping(Person::getLastName, toSet())));
* }</pre>
*
* @implNote
* The returned {@code Collector} is not concurrent. For parallel stream
* pipelines, the {@code combiner} function operates by merging the keys
* from one map into another, which can be an expensive operation. If
* preservation of the order in which elements are presented to the downstream
* collector is not required, using {@link #groupingByConcurrent(Function, Supplier, Collector)}
* may offer better parallel performance.
*
* @param <T> the type of the input elements
* @param <K> the type of the keys
* @param <A> the intermediate accumulation type of the downstream collector
* @param <D> the result type of the downstream reduction
* @param <M> the type of the resulting {@code Map}
* @param classifier a classifier function mapping input elements to keys
* @param downstream a {@code Collector} implementing the downstream reduction
* @param mapFactory a function which, when called, produces a new empty
* {@code Map} of the desired type
* @return a {@code Collector} implementing the cascaded group-by operation
*
* @see #groupingBy(Function, Collector)
* @see #groupingBy(Function)
* @see #groupingByConcurrent(Function, Supplier, Collector)
*/
public static <T, K, D, A, M extends Map<K, D>>
Collector<T, ?, M> groupingBy(Function<? super T, ? extends K> classifier,
Supplier<M> mapFactory,
Collector<? super T, A, D> downstream) {
Supplier<A> downstreamSupplier = downstream.supplier();
BiConsumer<A, ? super T> downstreamAccumulator = downstream.accumulator();
BiConsumer<Map<K, A>, T> accumulator = (m, t) -> {
K key = Objects.requireNonNull(classifier.apply(t), "element cannot be mapped to a null key");
A container = m.computeIfAbsent(key, k -> downstreamSupplier.get());
downstreamAccumulator.accept(container, t);
};
BinaryOperator<Map<K, A>> merger = Collectors.<K, A, Map<K, A>>mapMerger(downstream.combiner());
@SuppressWarnings("unchecked")
Supplier<Map<K, A>> mangledFactory = (Supplier<Map<K, A>>) mapFactory;
if (downstream.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)) {
return new CollectorImpl<>(mangledFactory, accumulator, merger, CH_ID);
}
else {
@SuppressWarnings("unchecked")
Function<A, A> downstreamFinisher = (Function<A, A>) downstream.finisher();
Function<Map<K, A>, M> finisher = intermediate -> {
intermediate.replaceAll((k, v) -> downstreamFinisher.apply(v));
@SuppressWarnings("unchecked")
M castResult = (M) intermediate;
return castResult;
};
return new CollectorImpl<>(mangledFactory, accumulator, merger, finisher, CH_NOID);
}
}
主要对比一下泛型参数吧,这里的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一下:
// Accumulate names into a List
List<String> list = people.stream().map(Person::getName).collect(Collectors.toList());
// Accumulate names into a TreeSet
Set<String> set = people.stream().map(Person::getName).collect(Collectors.toCollection(TreeSet::new));
// Convert elements to strings and concatenate them, separated by commas
String joined = things.stream()
.map(Object::toString)
.collect(Collectors.joining(", "));
// Compute sum of salaries of employee
int total = employees.stream()
.collect(Collectors.summingInt(Employee::getSalary)));
// Group employees by department
Map<Department, List<Employee>> byDept
= employees.stream()
.collect(Collectors.groupingBy(Employee::getDepartment));
// Compute sum of salaries by department
Map<Department, Integer> totalByDept
= employees.stream()
.collect(Collectors.groupingBy(Employee::getDepartment,
Collectors.summingInt(Employee::getSalary)));
// Partition students into passing and failing
Map<Boolean, List<Student>> passingFailing =
students.stream()
.collect(Collectors.partitioningBy(s -> s.getGrade() >= PASS_THRESHOLD));
我们拿一个简单的例子来详细分析一下:
String[] a = {"1.0", "2.0", "3.0", "4.0", "5.0"};
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);
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
我们之前已经讲过构造流源头的分析,产生了一个ReferencePipeline的Head。那么从后续的map开始分析:
public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {
Objects.requireNonNull(mapper);
return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
@Override
Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
return new Sink.ChainedReference<P_OUT, R>(sink) {
@Override
public void accept(P_OUT u) {
downstream.accept(mapper.apply(u));
}
};
}
};
}
仔细看,map返回的是一个StatelessOp,它继承了ReferencePipeline。我们知道ReferencePipeline继承了AbstractPipeline,而AbstractPipeline是真正流结构的持有者。AbstractPipeline里定义了source、nextStage和previousStage,这个结构很明确的将流等同于一个链表,每个node都和前后有关联,并且每个node都有一个到链表起始head的连接。如下图所示:
AbstractPipeline有个onWrapSink抽象方法,其核心就是将每个pipeline表达为一个Sink,而一个流就是由一个源头Source,多个Sink串起来的一个链表结构。每个Sink结构上作为一个链表节点存在,功能上等价于一个Consumer(Sink接口继承Consumer),通过accept方法来执行Sink操作。额外的多了begin和end方法做一些预处理和后处理的工作。
回到刚才的map方法,map返回一个Sink(具体是一个ChainedReference),其中accept做的工作就是调用mapper的apply方法,相当于执行mapper了。做一个拓扑图,就是
接下来构造filter,源码如下:
public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
Objects.requireNonNull(predicate);
return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
StreamOpFlag.NOT_SIZED) {
@Override
Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {
@Override
public void begin(long size) {
downstream.begin(-1);
}
@Override
public void accept(P_OUT u) {
if (predicate.test(u))
downstream.accept(u);
}
};
}
};
}
同理分析,它返回的也是个StatelessOp,其中同样是一个ChainedReference的 Sink,accept方法就是会判断一下谓词函数predicate是否接受参数,如果接受调用下游的accept。拓扑变更为
再接下来排序,排序返回的是一个StatefulOp,对比之前的map和filter,排序是一个有状态的中间操作。实际的StatefulOp是一个它的子类OfRef,OfRef持有一个排序器
private static final class OfRef<T> extends ReferencePipeline.StatefulOp<T, T> {
/**
* Comparator used for sorting
*/
private final boolean isNaturalSort;
private final Comparator<? super T> comparator;
...
}
它的onWrapSink方法返回
public Sink<T> opWrapSink(int flags, Sink<T> sink) {
Objects.requireNonNull(sink);
// If the input is already naturally sorted and this operation
// also naturally sorted then this is a no-op
if (StreamOpFlag.SORTED.isKnown(flags) && isNaturalSort)
return sink;
else if (StreamOpFlag.SIZED.isKnown(flags))
return new SizedRefSortingSink<>(sink, comparator);
else
return new RefSortingSink<>(sink, comparator);
}
也就是说,如果有序,直接pass,返回原sink,否则,根据其列表是否有界,来使用不同的Sink返回,有界的排序Sink内部是一个数组T[],而无界的排序Sink内部是一个ArrayList< T>。以无界排序Sink为例,其中的三个方法begin负责构造List,accept负责把流元素add进去,而end负责排序sort。看下实现一目了然:
private static final class RefSortingSink<T> extends AbstractRefSortingSink<T> {
private ArrayList<T> list;
RefSortingSink(Sink<? super T> sink, Comparator<? super T> comparator) {
super(sink, comparator);
}
@Override
public void begin(long size) {
if (size >= Nodes.MAX_ARRAY_SIZE)
throw new IllegalArgumentException(Nodes.BAD_SIZE);
list = (size >= 0) ? new ArrayList<T>((int) size) : new ArrayList<T>();
}
@Override
public void end() {
list.sort(comparator);
downstream.begin(list.size());
if (!cancellationWasRequested) {
list.forEach(downstream::accept);
}
else {
for (T t : list) {
if (downstream.cancellationRequested()) break;
downstream.accept(t);
}
}
downstream.end();
list = null;
}
@Override
public void accept(T t) {
list.add(t);
}
}
最终构造的Stream拓扑结构变为
limit也是一个StatefulOp,其和skip复用了一个Sink,其中的onWrapSink方法具体会在accept阶段控制一个limit的size:
public void accept(T t) {
if (n == 0) {
if (m > 0) {
m--;
downstream.accept(t);
}
}
else {
n--;
}
}
其他的逻辑和之前的sink类似,不赘述,拓扑结构更新如下:
reduce就和之前的Sink不一样了,reduce作为一个终止操作,不再是Sink,而是一个触发流执行的动作,其调用了AbstractPipeline的evaluate方法来触发整个流的执行。
public final Optional<P_OUT> reduce(BinaryOperator<P_OUT> accumulator) {
return evaluate(ReduceOps.makeRef(accumulator));
}
evaluate方法声明如下:
/**
* Evaluate the pipeline with a terminal operation to produce a result.
*
* @param <R> the type of result
* @param terminalOp the terminal operation to be applied to the pipeline.
* @return the result
*/
final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {
assert getOutputShape() == terminalOp.inputShape();
if (linkedOrConsumed)
throw new IllegalStateException(MSG_STREAM_LINKED);
linkedOrConsumed = true;
return isParallel()
? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
: terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
}
其入参是一个TerminalOp终止操作,而返回的就是流执行后的结果。干的事情就是调用TerminalOp的evaluateSequential方法来执行整个流。具体到这个例子就是调用ReduceOp(它继承TerminalOp)的evaluateSequential方法。
public <P_IN> R evaluateSequential(PipelineHelper<T> helper,
Spliterator<P_IN> spliterator) {
return helper.wrapAndCopyInto(makeSink(), spliterator).get();
}
其中PipelineHelper的wrapAndCopyInto方法就是包装sink,然后copyInto就是执行了。
final <P_IN, S extends Sink<E_OUT>> S wrapAndCopyInto(S sink, Spliterator<P_IN> spliterator) {
copyInto(wrapSink(Objects.requireNonNull(sink)), spliterator);
return sink;
}
其中wrapSink很重要,它负责调用之前拓扑结构的每个Sink的onWrapSink方法把这个链式结构建立起来。看代码会发现,链是自后向前的。
final <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) {
Objects.requireNonNull(sink);
for ( @SuppressWarnings("rawtypes") AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) {
sink = p.opWrapSink(p.previousStage.combinedFlags, sink);
}
return (Sink<P_IN>) sink;
}
构造好后,执行copyInto:
final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
Objects.requireNonNull(wrappedSink);
if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
wrappedSink.begin(spliterator.getExactSizeIfKnown());
spliterator.forEachRemaining(wrappedSink);
wrappedSink.end();
}
else {
copyIntoWithCancel(wrappedSink, spliterator);
}
}
具体核心就是驱动源头的Spliterator开始forEachRemaining,也就是遍历。对于每个元素进行tryAdvance(action),此处的action就是一个Sink了(刚才说过Sink继承Consumer)。这样一个链就开始从头到尾(不一定,这取决于源头结构和Spliterator类型)执行了。
最终的链式拓扑结构如下:
链式的执行依赖于TerminalOp究竟是什么,因此一个Stream的执行是lazy的,当流构建好时,只是一个Sink的链式结构,最终的遍历和执行需要一个终止操作来触发。回到本节开始时的例子,前5步流是构造不执行的,直到第6步确定reduce求和,才触发了流的遍历执行。
本文是Java8三个系列的最后一篇,笔者非常粗浅地分析了Java 8的Stream用法和简单实现剖析。整个一个系列完成,个人以为Java 8的三大亮点也分别简单总结整理完成。对于Java开发者,个人建议要在程序设计时广泛使用流式Stream用法,其可以非常高效地帮助开发者完成一些常用结构体(array/list/map)的遍历操作;使用函数式设计,可以极大的补充面向对象的编程思维,在OO模式抽象实体的同时,可以FP模式抽象操作和函数,这对于一些以操作为主的系统架构和设计来说,引入方法抽象和函数将极大地辅助系统设计。