山东省住房城乡建设厅网站首页,wordpress菜单背景半透明,开放平台api,网站运营系统前面我们已经学会如何使用Stream API#xff0c;用起来真的很爽#xff0c;但简洁的方法下面似乎隐藏着无尽的秘密#xff0c;如此强大的API是如何实现的呢#xff1f;比如Pipeline是怎么执行的#xff0c;每次方法调用都会导致一次迭代吗#xff1f;自动并行又是怎么做到…前面我们已经学会如何使用Stream API用起来真的很爽但简洁的方法下面似乎隐藏着无尽的秘密如此强大的API是如何实现的呢比如Pipeline是怎么执行的每次方法调用都会导致一次迭代吗自动并行又是怎么做到的线程个数是多少本节我们学习Stream流水线的原理这是Stream实现的关键所在。
首先回顾一下容器执行Lambda表达式的方式以ArrayList.forEach()方法为例具体代码如下
// ArrayList.forEach()
public void forEach(Consumer? super E action) {...for (int i0; modCount expectedModCount i size; i) {action.accept(elementData[i]);// 回调方法}...
}
我们看到ArrayList.forEach()方法的主要逻辑就是一个for循环在该for循环里不断调用action.accept()回调方法完成对元素的遍历。这完全没有什么新奇之处回调方法在Java GUI的监听器中广泛使用。Lambda表达式的作用就是相当于一个回调方法这很好理解。
Stream API中大量使用Lambda表达式作为回调方法但这并不是关键。理解Stream我们更关心的是另外两个问题流水线和自动并行。使用Stream或许很容易写入如下形式的代码
int longestStringLengthStartingWithA strings.stream().filter(s - s.startsWith(A)).mapToInt(String::length).max();
上述代码求出以字母A开头的字符串的最大长度一种直白的方式是为每一次函数调用都执一次迭代这样做能够实现功能但效率上肯定是无法接受的。类库的实现着使用流水线Pipeline的方式巧妙的避免了多次迭代其基本思想是在一次迭代中尽可能多的执行用户指定的操作。为讲解方便我们汇总了Stream的所有操作。
Stream操作分类中间操作(Intermediate operations)无状态(Stateless)unordered() filter() map() mapToInt() mapToLong() mapToDouble() flatMap() flatMapToInt() flatMapToLong() flatMapToDouble() peek()有状态(Stateful)distinct() sorted() sorted() limit() skip()结束操作(Terminal operations)非短路操作forEach() forEachOrdered() toArray() reduce() collect() max() min() count()短路操作(short-circuiting)anyMatch() allMatch() noneMatch() findFirst() findAny()
Stream上的所有操作分为两类中间操作和结束操作中间操作只是一种标记只有结束操作才会触发实际计算。中间操作又可以分为无状态的(Stateless)和有状态的(Stateful)无状态中间操作是指元素的处理不受前面元素的影响而有状态的中间操作必须等到所有元素处理之后才知道最终结果比如排序是有状态操作在读取所有元素之前并不能确定排序结果结束操作又可以分为短路操作和非短路操作短路操作是指不用处理全部元素就可以返回结果比如找到第一个满足条件的元素。之所以要进行如此精细的划分是因为底层对每一种情况的处理方式不同。
一种直白的实现方式 仍然考虑上述求最长字符串的程序一种直白的流水线实现方式是为每一次函数调用都执一次迭代并将处理中间结果放到某种数据结构中比如数组容器等。具体说来就是调用filter()方法后立即执行选出所有以A开头的字符串并放到一个列表list1中之后让list1传递给mapToInt()方法并立即执行生成的结果放到list2中最后遍历list2找出最大的数字作为最终结果。程序的执行流程如如所示
这样做实现起来非常简单直观但有两个明显的弊端 迭代次数多。迭代次数跟函数调用的次数相等。 频繁产生中间结果。每次函数调用都产生一次中间结果存储开销无法接受。
这些弊端使得效率底下根本无法接受。如果不使用Stream API我们都知道上述代码该如何在一次迭代中完成大致是如下形式
int longest 0;
for(String str : strings){if(str.startsWith(A)){// 1. filter(), 保留以A开头的字符串int len str.length();// 2. mapToInt(), 转换成长度longest Math.max(len, longest);// 3. max(), 保留最长的长度}
}
采用这种方式我们不但减少了迭代次数也避免了存储中间结果显然这就是流水线因为我们把三个操作放在了一次迭代当中。只要我们事先知道用户意图总是能够采用上述方式实现跟Stream API等价的功能但问题是Stream类库的设计者并不知道用户的意图是什么。如何在无法假设用户行为的前提下实现流水线是类库的设计者要考虑的问题。 Stream流水线解决方案
我们大致能够想到应该采用某种方式记录用户每一步的操作当用户调用结束操作时将之前记录的操作叠加到一起在一次迭代中全部执行掉。沿着这个思路有几个问题需要解决 用户的操作如何记录 操作如何叠加 叠加之后的操作如何执行 执行后的结果如果有在哪里
操作如何记录 注意这里使用的是“操作(operation)”一词指的是“Stream中间操作”的操作很多Stream操作会需要一个回调函数Lambda表达式因此一个完整的操作是数据来源操作回调函数构成的三元组。Stream中使用Stage的概念来描述一个完整的操作并用某种实例化后的PipelineHelper来代表Stage将具有先后顺序的各个Stage连到一起就构成了整个流水线。跟Stream相关类和接口的继承关系图示。
还有IntPipeline, LongPipeline, DoublePipeline没在图中画出这三个类专门为三种基本类型不是包装类型而定制的跟ReferencePipeline是并列关系。图中Head用于表示第一个Stage即调用调用诸如Collection.stream()方法产生的Stage很显然这个Stage里不包含任何操作StatelessOp和StatefulOp分别表示无状态和有状态的Stage对应于无状态和有状态的中间操作。
Stream流水线组织结构示意图如下 图中通过Collection.stream()方法得到Head也就是stage0紧接着调用一系列的中间操作不断产生新的Stream。这些Stream对象以双向链表的形式组织在一起构成整个流水线由于每个Stage都记录了前一个Stage和本次的操作以及回调函数依靠这种结构就能建立起对数据源的所有操作。这就是Stream记录操作的方式。
操作如何叠加
以上只是解决了操作记录的问题要想让流水线起到应有的作用我们需要一种将所有操作叠加到一起的方案。你可能会觉得这很简单只需要从流水线的head开始依次执行每一步的操作包括回调函数就行了。这听起来似乎是可行的但是你忽略了前面的Stage并不知道后面Stage到底执行了哪种操作以及回调函数是哪种形式。换句话说只有当前Stage本身才知道该如何执行自己包含的动作。这就需要有某种协议来协调相邻Stage之间的调用关系。
这种协议由Sink接口完成Sink接口包含的方法如下表所示 方法名作用void begin(long size)开始遍历元素之前调用该方法通知Sink做好准备。void end()所有元素遍历完成之后调用通知Sink没有更多的元素了。boolean cancellationRequested()是否可以结束操作可以让短路操作尽早结束。void accept(T t)遍历元素时调用接受一个待处理元素并对元素进行处理。Stage把自己包含的操作和回调方法封装到该方法里前一个Stage只需要调用当前Stage.accept(T t)方法就行了。
有了上面的协议相邻Stage之间调用就很方便了每个Stage都会将自己的操作封装到一个Sink里前一个Stage只需调用后一个Stage的accept()方法即可并不需要知道其内部是如何处理的。当然对于有状态的操作Sink的begin()和end()方法也是必须实现的。比如Stream.sorted()是一个有状态的中间操作其对应的Sink.begin()方法可能创建一个乘放结果的容器而accept()方法负责将元素添加到该容器最后end()负责对容器进行排序。对于短路操作Sink.cancellationRequested()也是必须实现的比如Stream.findFirst()是短路操作只要找到一个元素cancellationRequested()就应该返回true以便调用者尽快结束查找。Sink的四个接口方法常常相互协作共同完成计算任务。实际上Stream API内部实现的的本质就是如何重载Sink的这四个接口方法。
有了Sink对操作的包装Stage之间的调用问题就解决了执行时只需要从流水线的head开始对数据源依次调用每个Stage对应的Sink.{begin(), accept(), cancellationRequested(), end()}方法就可以了。一种可能的Sink.accept()方法流程是这样的
void accept(U u){1. 使用当前Sink包装的回调函数处理u2. 将处理结果传递给流水线下游的Sink
}
Sink接口的其他几个方法也是按照这种[处理-转发]的模型实现。下面我们结合具体例子看看Stream的中间操作是如何将自身的操作包装成Sink以及Sink是如何将处理结果转发给下一个Sink的。先看Stream.map()方法
// Stream.map()调用该方法将产生一个新的Stream
public final R StreamR map(Function? super P_OUT, ? extends R mapper) {...return new StatelessOpP_OUT, R(this, StreamShape.REFERENCE,StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {Override /*opWripSink()方法返回由回调函数包装而成Sink*/SinkP_OUT opWrapSink(int flags, SinkR downstream) {return new Sink.ChainedReferenceP_OUT, R(downstream) {Overridepublic void accept(P_OUT u) {R r mapper.apply(u);// 1. 使用当前Sink包装的回调函数mapper处理udownstream.accept(r);// 2. 将处理结果传递给流水线下游的Sink}};}};
}
上述代码看似复杂其实逻辑很简单就是将回调函数mapper包装到一个Sink当中。由于Stream.map()是一个无状态的中间操作所以map()方法返回了一个StatelessOp内部类对象一个新的Stream调用这个新Stream的opWripSink()方法将得到一个包装了当前回调函数的Sink。
再来看一个复杂一点的例子。Stream.sorted()方法将对Stream中的元素进行排序显然这是一个有状态的中间操作因为读取所有元素之前是没法得到最终顺序的。抛开模板代码直接进入问题本质sorted()方法是如何将操作封装成Sink的呢sorted()一种可能封装的Sink代码如下
// Stream.sort()方法用到的Sink实现
class RefSortingSinkT extends AbstractRefSortingSinkT {private ArrayListT list;// 存放用于排序的元素RefSortingSink(Sink? super T downstream, Comparator? super T comparator) {super(downstream, comparator);}Overridepublic void begin(long size) {...// 创建一个存放排序元素的列表list (size 0) ? new ArrayListT((int) size) : new ArrayListT();}Overridepublic void end() {list.sort(comparator);// 只有元素全部接收之后才能开始排序downstream.begin(list.size());if (!cancellationWasRequested) {// 下游Sink不包含短路操作list.forEach(downstream::accept);// 2. 将处理结果传递给流水线下游的Sink}else {// 下游Sink包含短路操作for (T t : list) {// 每次都调用cancellationRequested()询问是否可以结束处理。if (downstream.cancellationRequested()) break;downstream.accept(t);// 2. 将处理结果传递给流水线下游的Sink}}downstream.end();list null;}Overridepublic void accept(T t) {list.add(t);// 1. 使用当前Sink包装动作处理t只是简单的将元素添加到中间列表当中}
}
上述代码完美的展现了Sink的四个接口方法是如何协同工作的 首先beging()方法告诉Sink参与排序的元素个数方便确定中间结果容器的的大小 之后通过accept()方法将元素添加到中间结果当中最终执行时调用者会不断调用该方法直到遍历所有元素 最后end()方法告诉Sink所有元素遍历完毕启动排序步骤排序完成后将结果传递给下游的Sink 如果下游的Sink是短路操作将结果传递给下游时不断询问下游cancellationRequested()是否可以结束处理。
叠加之后的操作如何执行 Sink完美封装了Stream每一步操作并给出了[处理-转发]的模式来叠加操作。这一连串的齿轮已经咬合就差最后一步拨动齿轮启动执行。是什么启动这一连串的操作呢也许你已经想到了启动的原始动力就是结束操作(Terminal Operation)一旦调用某个结束操作就会触发整个流水线的执行。
结束操作之后不能再有别的操作所以结束操作不会创建新的流水线阶段(Stage)直观的说就是流水线的链表不会在往后延伸了。结束操作会创建一个包装了自己操作的Sink这也是流水线中最后一个Sink这个Sink只需要处理数据而不需要将结果传递给下游的Sink因为没有下游。对于Sink的[处理-转发]模型结束操作的Sink就是调用链的出口。
我们再来考察一下上游的Sink是如何找到下游Sink的。一种可选的方案是在PipelineHelper中设置一个Sink字段在流水线中找到下游Stage并访问Sink字段即可。但Stream类库的设计者没有这么做而是设置了一个Sink AbstractPipeline.opWrapSink(int flags, Sink downstream)方法来得到Sink该方法的作用是返回一个新的包含了当前Stage代表的操作以及能够将结果传递给downstream的Sink对象。为什么要产生一个新对象而不是返回一个Sink字段这是因为使用opWrapSink()可以将当前操作与下游Sink上文中的downstream参数结合成新Sink。试想只要从流水线的最后一个Stage开始不断调用上一个Stage的opWrapSink()方法直到最开始不包括stage0因为stage0代表数据源不包含操作就可以得到一个代表了流水线上所有操作的Sink用代码表示就是这样
// AbstractPipeline.wrapSink()
// 从下游向上游不断包装Sink。如果最初传入的sink代表结束操作
// 函数返回时就可以得到一个代表了流水线上所有操作的Sink。
final P_IN SinkP_IN wrapSink(SinkE_OUT sink) {...for (AbstractPipeline pAbstractPipeline.this; p.depth 0; pp.previousStage) {sink p.opWrapSink(p.previousStage.combinedFlags, sink);}return (SinkP_IN) sink;
}
现在流水线上从开始到结束的所有的操作都被包装到了一个Sink里执行这个Sink就相当于执行整个流水线执行Sink的代码如下
// AbstractPipeline.copyInto(), 对spliterator代表的数据执行wrappedSink代表的操作。
final P_IN void copyInto(SinkP_IN wrappedSink, SpliteratorP_IN spliterator) {...if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {wrappedSink.begin(spliterator.getExactSizeIfKnown());// 通知开始遍历spliterator.forEachRemaining(wrappedSink);// 迭代wrappedSink.end();// 通知遍历结束}...
}
上述代码首先调用wrappedSink.begin()方法告诉Sink数据即将到来然后调用spliterator.forEachRemaining()方法对数据进行迭代Spliterator是容器的一种迭代器参阅最后调用wrappedSink.end()方法通知Sink数据处理结束。逻辑如此清晰。
执行后的结果在哪里
最后一个问题是流水线上所有操作都执行后用户所需要的结果如果有在哪里首先要说明的是不是所有的Stream结束操作都需要返回结果有些操作只是为了使用其副作用(Side-effects)比如使用Stream.forEach()方法将结果打印出来就是常见的使用副作用的场景事实上除了打印之外其他场景都应避免使用副作用对于真正需要返回结果的结束操作结果存在哪里呢 特别说明副作用不应该被滥用也许你会觉得在Stream.forEach()里进行元素收集是个不错的选择就像下面代码中那样但遗憾的是这样使用的正确性和效率都无法保证因为Stream可能会并行执行。大多数使用副作用的地方都可以使用归约操作更安全和有效的完成。 // 错误的收集方式
ArrayListString results new ArrayList();
stream.filter(s - pattern.matcher(s).matches()).forEach(s - results.add(s)); // Unnecessary use of side-effects!
// 正确的收集方式
ListStringresults stream.filter(s - pattern.matcher(s).matches()).collect(Collectors.toList()); // No side-effects!
回到流水线执行结果的问题上来需要返回结果的流水线结果存在哪里呢这要分不同的情况讨论下表给出了各种有返回结果的Stream结束操作。
返回类型对应的结束操作booleananyMatch() allMatch() noneMatch()OptionalfindFirst() findAny()归约结果reduce() collect()数组toArray()对于表中返回boolean或者Optional的操作Optional是存放 一个 值的容器的操作由于值返回一个值只需要在对应的Sink中记录这个值等到执行结束时返回就可以了。 对于归约操作最终结果放在用户调用时指定的容器中容器类型通过收集器指定。collect(), reduce(), max(), min()都是归约操作虽然max()和min()也是返回一个Optional但事实上底层是通过调用reduce()方法实现的。 对于返回是数组的情况毫无疑问的结果会放在数组当中。这么说当然是对的但在最终返回数组之前结果其实是存储在一种叫做Node的数据结构中的。Node是一种多叉树结构元素存储在树的叶子当中并且一个叶子节点可以存放多个元素。这样做是为了并行执行方便。关于Node的具体结构我们会在下一节探究Stream如何并行执行时给出详细说明。
本文详细介绍了Stream流水线的组织方式和执行过程学习本文将有助于理解原理并写出正确的Stream代码同时打消你对Stream API效率方面的顾虑。如你所见Stream API实现如此巧妙即使我们使用外部迭代手动编写等价代码也未必更加高效。
