学习Java8的Stream-创新互联
Stream把对一个集合的很多操作,以一种流式作业串起来,按照类似函数式编程书写代码,界面清爽清晰。
Stream不同于Guava的FluentIterable系列。FluentIterable系列,是通过封装集合(List、Set等)并且重载迭代器、get的方式,进行的transform、filter等,优点是简单并且性能高。缺点是功能单一、并且容易误用。比如,对transform之后的列表的每个项,本质上都是一个视图(View),而不是实际的项(值对象)。多次调用get方法。这本质上每次都会层层调用Function的apply方法,如果其中有复杂运算或者I/O读取,效率是非常低的。
Stream本身,建立了一系列的数据结构,其中的数据都可以看成是实际存在的数据,而并不仅仅是视图。网上分析源代码的文章非常多,这里不重复描述,而是仅仅介绍理解的一个思路。
首先,是几个关键的数据结构。
Stream。接口是BaseStream,可以看做是流基本『形态』的描述。或者说,每个流式作业。虽然后面提到的Pipeline也是实现了BaseStream,但是这样似乎更容易理解。
Pipeline。公共父类是AbstractPipeline。可以看做是流式作业中的每个作业,或者说是每个作业节点。为了防止不必要的装箱和拆箱操作,又分成了ReferencePipeline、IntPipeline、LongPipeline、DoublePipeline。这些管道可以分成三种:头、终结操作、非终结操作。分别对应的类是XXXXPipeline.Head、XXXXPipeline.StatelessOp和XXXXPipeline.StatefulOp、TerminalOp.调用流的工作方法,都会制造出来这样一个Pipeline。比如,调用IntStream.of(),就会生成一个头;调用Stream.map(),会生成一个非终结操作;调用Stream.reduce(),生成一个终结操作。非终结操作里面,都要实现opWrapSink方法,该方法要返回一个Sink。
Sink。每个操作,对应一个Sink。每个Sink,关注三个方法:begin、end、accept。如果当前的Sink没有操作,那么直接调用downstream.accept;否则,把当前操作结果作为参数调用downstream.accept。downstream是什么概念呢?比如,一个流的操作是IntStream.of(1,2,3).map(x -> x + 1).boxed().max()。那么map对应的sink,就是头的downstream,boxed对应的sink就是map的downstream。综合起来,只要调用了头的accept,就会层层调用到最后一个终结操作。终结操作没有opWrapSink方法,所以自然不会调用到后续的流。
Spliterator。流里面数据的访问工具。如果是串行流,一般是直接调用里面的forEachRemaining。该方法里关注action.accept,如果之前串联好了每个Sink,那么这里一句调用,就开始了Sink的层层调用。
至此,基本数据结构就介绍这些。下面,关注流的每个节点,是如何连起来的。
每个Stream,由一组Pipeline节点组成,每追加一个操作,都会向这组Pipeline后面追加一个Pipeline结构。追加时候维护的信息里面,关注sourceStage(头),previousStage(上一个Pipeline),sourceSupplier(流数据的Spliterator),depth(Pipeline长度)。一直到最后的终结操作,连城一个Pipeline链。
对于终结操作,不论是reduce,还是collect操作,都会调用到AbstracePipeline.evaluate方法。以串行流的reduce为例,直接调用到AbstractPipeline.wrapAndCopyInto。
其中,wrap是遍历Pipeline链,调用每个阶段的opWrapSink。这样每个Sink通过方法逐层调用(而非内存数据上的指针链接),从第一个Pipeline的Sink链接到末尾。
@Override
@SuppressWarnings("unchecked")
final Sink wrapSink(Sink 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) sink;
}
copyInto是依次调用Sink.begin,Spliterator.forEachRemaing,Sink.end方法。上文曾经提到,forEachRemaing会以流里的每项数据为参数,层层调用每级Sink的accept。
@Override
final void copyInto(Sink wrappedSink, Spliterator spliterator) {
Objects.requireNonNull(wrappedSink);
if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
wrappedSink.begin(spliterator.getExactSizeIfKnown());
spliterator.forEachRemaining(wrappedSink);
wrappedSink.end();
}
else {
copyIntoWithCancel(wrappedSink, spliterator);
}
}
至此,整个流的操作,自上而下完全联系到了一起。
创新互联www.cdcxhl.cn,专业提供香港、美国云服务器,动态BGP最优骨干路由自动选择,持续稳定高效的网络助力业务部署。公司持有工信部办法的idc、isp许可证, 机房独有T级流量清洗系统配攻击溯源,准确进行流量调度,确保服务器高可用性。佳节活动现已开启,新人活动云服务器买多久送多久。
标题名称:学习Java8的Stream-创新互联
网址分享:http://scjbc.cn/article/edgcs.html