06讲Stream如何提⾼遍历集合效率
你好，我是刘超。
上⼀讲中，我在讲List集合类，那我想你⼀定也知道集合的顶端接⼝Collection。在Java8中，Collection新增了两个流⽅法，分别是Stream()和parallelStream()。
通过英⽂名不难猜测，这两个⽅法肯定和Stream有关，那进⼀步猜测，是不是和我们熟悉的InputStream和OutputStream也有关系呢？集合类中新增的两个Stream⽅法到底有什么作⽤？今天，我们就来深⼊了解下Stream。
什么是Stream？
现在很多⼤数据量系统中都存在分表分库的情况。
例如，电商系统中的订单表，常常使⽤⽤户ID的Hash值来实现分表分库，这样是为了减少单个表的数据量，优化⽤户查询订单的速度。
但在后台管理员审核订单时，他们需要将各个数据源的数据查询到应⽤层之后进⾏合并操作。
例如，当我们需要查询出过滤条件下的所有订单，并按照订单的某个条件进⾏排序，单个数据源查询出来的数据是可以按照某个条件进⾏排序的，但多个数据源查询出来已经排序好的数据，并不代表合并后是正确的排序，所以我们需要在应⽤层对合并数据集合重新进⾏排序。
在Java8之前，我们通常是通过for循环或者Iterator迭代来重新排序合并数据，⼜或者通过重新定义Collections.sorts的
Comparator⽅法来实现，这两种⽅式对于⼤数据量系统来说，效率并不是很理想。
Java8中添加了⼀个新的接⼝类Stream，他和我们之前接触的字节流概念不太⼀样，Java8集合中的Stream相当于⾼级版的
Iterator，他可以通过Lambda 表达式对集合进⾏各种⾮常便利、⾼效的聚合操作（Aggregate Operation），或者⼤批量数据操作 (Bulk Data Operation)。
Stream的聚合操作与数据库SQL的聚合操作sorted、filter、map等类似。我们在应⽤层就可以⾼效地实现类似数据库SQL的聚
合操作了，⽽在数据操作⽅⾯，Stream不仅可以通过串⾏的⽅式实现数据操作，还可以通过并⾏的⽅式处理⼤批量数据，提
⾼数据的处理效率。
接下来我们就⽤⼀个简单的例⼦来体验下Stream的简洁与强⼤。
这个Demo的需求是过滤分组⼀所中学⾥身⾼在160cm以上的男⼥同学，我们先⽤传统的迭代⽅式来实现，代码如下：
Map if (stu.getHeight() > 160) { //如果身⾼⼤于160 if (stuMap.get(stu.getSex()) == null) { //该性别还没分类 List list.add(stu);//将学⽣放进去列表stuMap.put(stu.getSex(), list);//将列表放到map中 } else { //该性别分类已存在 stuMap.get(stu.getSex()).add(stu);//该性别分类已存在，则直接放进去即可 } } } |
|---|
我们再使⽤Java8中的Stream API进⾏实现：
1.串⾏实现
Map |
|---|
 |
2.并⾏实现
Map |
|---|
 |
通过上⾯两个简单的例⼦，我们可以发现，Stream结合Lambda表达式实现遍历筛选功能⾮常得简洁和便捷。
Stream如何优化遍历？
上⾯我们初步了解了Java8中的Stream API，那Stream是如何做到优化迭代的呢？并⾏⼜是如何实现的？下⾯我们就透过
Stream源码剖析Stream的实现原理。
Stream操作分类
在了解Stream的实现原理之前，我们先来了解下Stream的操作分类，因为他的操作分类其实是实现⾼效迭代⼤数据集合的重要原因之⼀。为什么这样说，分析完你就清楚了。
官⽅将Stream中的操作分为两⼤类：中间操作（Intermediate operations）和终结操作（Terminal operations）。中间操作只对操作进⾏了记录，即只会返回⼀个流，不会进⾏计算操作，⽽终结操作是实现了计算操作。
中间操作⼜可以分为⽆状态（Stateless）与有状态（Stateful）操作，前者是指元素的处理不受之前元素的影响，后者是指该
操作只有拿到所有元素之后才能继续下去。
终结操作⼜可以分为短路（Short-circuiting）与⾮短路（Unshort-circuiting）操作，前者是指遇到某些符合条件的元素就可以得到最终结果，后者是指必须处理完所有元素才能得到最终结果。操作分类详情如下图所示：
我们通常还会将中间操作称为懒操作，也正是由这种懒操作结合终结操作、数据源构成的处理管道（Pipeline），实现了
Stream的⾼效。
Stream源码实现
在了解Stream如何⼯作之前，我们先来了解下Stream包是由哪些主要结构类组合⽽成的，各个类的职责是什么。参照下图：
BaseStream和Stream为最顶端的接⼝类。BaseStream主要定义了流的基本接⼝⽅法，例如，spliterator、isParallel等；
Stream则定义了⼀些流的常⽤操作⽅法，例如，map、filter等。
ReferencePipeline是⼀个结构类，他通过定义内部类组装了各种操作流。他定义了Head、StatelessOp、StatefulOp三个内部类，实现了BaseStream与Stream的接⼝⽅法。
Sink接⼝是定义每个Stream操作之间关系的协议，他包含begin()、end()、cancellationRequested()、accpt()四个⽅法。
ReferencePipeline最终会将整个Stream流操作组装成⼀个调⽤链，⽽这条调⽤链上的各个Stream操作的上下关系就是通过
Sink接⼝协议来定义实现的。
Stream操作叠加
我们知道，⼀个Stream的各个操作是由处理管道组装，并统⼀完成数据处理的。在JDK中每次的中断操作会以使⽤阶段
（Stage）命名。
管道结构通常是由ReferencePipeline类实现的，前⾯讲解Stream包结构时，我提到过ReferencePipeline包含了Head、
StatelessOp、StatefulOp三种内部类。
Head类主要⽤来定义数据源操作，在我们初次调⽤names.stream()⽅法时，会初次加载Head对象，此时为加载数据源操作； 接着加载的是中间操作，分别为⽆状态中间操作StatelessOp对象和有状态操作StatefulOp对象，此时的Stage并没有执⾏，⽽是通过AbstractPipeline⽣成了⼀个中间操作Stage链表；当我们调⽤终结操作时，会⽣成⼀个最终的Stage，通过这个Stage 触发之前的中间操作，从最后⼀个Stage开始，递归产⽣⼀个Sink链。如下图所示：
下⾯我们再通过⼀个例⼦来感受下Stream的操作分类是如何实现⾼效迭代⼤数据集合的。
List String maxLenStartWithZ = names.stream() .filter(name -> name.startsWith(“张”)) .mapToInt(String::length) .max() .toString(); |
|---|
这个例⼦的需求是查找出⼀个⻓度最⻓，并且以张为姓⽒的名字。从代码⻆度来看，你可能会认为是这样的操作流程：⾸先遍
历⼀次集合，得到以“张”开头的所有名字；然后遍历⼀次filter得到的集合，将名字转换成数字⻓度；最后再从⻓度集合中找到最⻓的那个名字并且返回。
这⾥我要很明确地告诉你，实际情况并⾮如此。我们来逐步分析下这个⽅法⾥所有的操作是如何执⾏的。
⾸先 ，因为names是ArrayList集合，所以names.stream()⽅法将会调⽤集合类基础接⼝Collection的Stream⽅法：
default Stream return StreamSupport.stream(spliterator(), false); } |
|---|
然后，Stream⽅法就会调⽤StreamSupport类的Stream⽅法，⽅法中初始化了⼀个ReferencePipeline的Head内部类对象：
public static return new ReferencePipeline.Head<>(spliterator, StreamOpFlag.fromCharacteristics(spliterator), parallel); } |
|---|
再调⽤filter和map⽅法，这两个⽅法都是⽆状态的中间操作，所以执⾏filter和map操作时，并没有进⾏任何的操作，⽽是分别创建了⼀个Stage来标识⽤户的每⼀次操作。
⽽通常情况下Stream的操作⼜需要⼀个回调函数，所以⼀个完整的Stage是由数据来源、操作、回调函数组成的三元组来表示。如下图所示，分别是ReferencePipeline的filter⽅法和map⽅法：
@Override public final Stream return new StatelessOp StreamOpFlag.NOT_SIZED) { @Override Sink @Override public void begin(long size) { downstream.begin(-1); } @Override public void accept(P_OUT u) { if (predicate.test(u)) downstream.accept(u); } }; } }; } |
|---|
@Override @SuppressWarnings(“unchecked”) public final return new StatelessOp StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) { @Override Sink @Override public void accept(P_OUT u) { downstream.accept(mapper.apply(u)); } }; } }; } |
|---|
new StatelessOp将会调⽤⽗类AbstractPipeline的构造函数，这个构造函数将前后的Stage联系起来，⽣成⼀个Stage链表：
AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) { if (previousStage.linkedOrConsumed) throw new IllegalStateException(MSG_STREAM_LINKED); previousStage.linkedOrConsumed = true; previousStage.nextStage = this;//将当前的stage的next指针指向之前的stage this.previousStage = previousStage;//赋值当前stage当全局变量previousStage this.sourceOrOpFlags = opFlags & StreamOpFlag.OP_MASK; this.combinedFlags = StreamOpFlag.combineOpFlags(opFlags, previousStage.combinedFlags); this.sourceStage = previousStage.sourceStage; if (opIsStateful()) sourceStage.sourceAnyStateful = true; this.depth = previousStage.depth + 1; } |
|---|
因为在创建每⼀个Stage时，都会包含⼀个opWrapSink()⽅法，该⽅法会把⼀个操作的具体实现封装在Sink类中，Sink采⽤
（处理->转发）的模式来叠加操作。
当执⾏max⽅法时，会调⽤ReferencePipeline的max⽅法，此时由于max⽅法是终结操作，所以会创建⼀个TerminalOp操作， 同时创建⼀个ReducingSink，并且将操作封装在Sink类中。
@Override public final Optional } |
|---|
最后，调⽤AbstractPipeline的wrapSink⽅法，该⽅法会调⽤opWrapSink⽣成⼀个Sink链表，Sink链表中的每⼀个Sink都封装了⼀个操作的具体实现。
@Override @SuppressWarnings(“unchecked”) final for ( @SuppressWarnings(“rawtypes”) AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previou sink = p.opWrapSink(p.previousStage.combinedFlags, sink); } return (Sink } |
|---|
 |
当Sink链表⽣成完成后，Stream开始执⾏，通过spliterator迭代集合，执⾏Sink链表中的具体操作。
@Override final if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) { wrappedSink.begin(spliterator.getExactSizeIfKnown()); spliterator.forEachRemaining(wrappedSink); wrappedSink.end(); } else { copyIntoWithCancel(wrappedSink, spliterator); } } |
|---|
Java8中的Spliterator的forEachRemaining会迭代集合，每迭代⼀次，都会执⾏⼀次filter操作，如果filter操作通过，就会触发
map操作，然后将结果放⼊到临时数组object中，再进⾏下⼀次的迭代。完成中间操作后，就会触发终结操作max。这就是串⾏处理⽅式了，那么Stream的另⼀种处理数据的⽅式⼜是怎么操作的呢？
Stream并⾏处理
Stream处理数据的⽅式有两种，串⾏处理和并⾏处理。要实现并⾏处理，我们只需要在例⼦的代码中新增⼀个Parallel()⽅法，代码如下所示：
List String maxLenStartWithZ = names.stream() .parallel() .filter(name -> name.startsWith(“张”)) .mapToInt(String::length) .max() .toString(); |
|---|
Stream的并⾏处理在执⾏终结操作之前，跟串⾏处理的实现是⼀样的。⽽在调⽤终结⽅法之后，实现的⽅式就有点不太⼀样，会调⽤TerminalOp的evaluateParallel⽅法进⾏并⾏处理。
final throw new IllegalStateException(MSG_STREAM_LINKED); linkedOrConsumed = true; return isParallel() ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags())) : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags())); } |
|---|
这⾥的并⾏处理指的是，Stream结合了ForkJoin框架，对Stream 处理进⾏了分⽚，Splititerator中的estimateSize⽅法会估算
出分⽚的数据量。
ForkJoin框架和估算算法，在这⾥我就不具体讲解了，如果感兴趣，你可以深⼊源码分析下该算法的实现。
通过预估的数据量获取最⼩处理单元的阈值，如果当前分⽚⼤⼩⼤于最⼩处理单元的阈值，就继续切分集合。每个分⽚将会⽣成⼀个Sink链表，当所有的分⽚操作完成后，ForkJoin框架将会合并分⽚任何结果集。
合理使⽤Stream
看到这⾥，你应该对Stream API是如何优化集合遍历有个清晰的认知了。Stream API⽤起来简洁，还能并⾏处理，那是不是使⽤Stream API，系统性能就更好呢？通过⼀组测试，我们⼀探究竟。
我们将对常规的迭代、Stream串⾏迭代以及Stream并⾏迭代进⾏性能测试对⽐，迭代循环中，我们将对数据进⾏过滤、分组等操作。分别进⾏以下⼏组测试：
多核CPU服务器配置环境下，对⽐⻓度100的int数组的性能；
多核CPU服务器配置环境下，对⽐⻓度1.00E+8的int数组的性能；
多核CPU服务器配置环境下，对⽐⻓度1.00E+8对象数组过滤分组的性能； 单核CPU服务器配置环境下，对⽐⻓度1.00E+8对象数组过滤分组的性能。
由于篇幅有限，我这⾥直接给出统计结果，你也可以⾃⼰去验证⼀下，具体的测试代码可以在Github上查看。通过以上测试， 我统计出的测试结果如下（迭代使⽤时间）：
常规的迭代
通过以上测试结果，我们可以看到：在循环迭代次数较少的情况下，常规的迭代⽅式性能反⽽更好；在单核CPU服务器配置环境中，也是常规迭代⽅式更有优势；⽽在⼤数据循环迭代中，如果服务器是多核CPU的情况下，Stream的并⾏迭代优势明显。所以我们在平时处理⼤数据的集合时，应该尽量考虑将应⽤部署在多核CPU环境下，并且使⽤Stream的并⾏迭代⽅式进
⾏处理。
⽤事实说话，我们看到其实使⽤Stream未必可以使系统性能更佳，还是要结合应⽤场景进⾏选择，也就是合理地使⽤
Stream。
总结
纵观Stream的设计实现，⾮常值得我们学习。从⼤的设计⽅向上来说，Stream将整个操作分解为了链式结构，不仅简化了遍历操作，还为实现了并⾏计算打下了基础。
从⼩的分类⽅向上来说，Stream将遍历元素的操作和对元素的计算分为中间操作和终结操作，⽽中间操作⼜根据元素之间状态有⽆⼲扰分为有状态和⽆状态操作，实现了链结构中的不同阶段。
在串⾏处理操作中，Stream在执⾏每⼀步中间操作时，并不会做实际的数据操作处理，⽽是将这些中间操作串联起来，最终 由终结操作触发，⽣成⼀个数据处理链表，通过Java8中的Spliterator迭代器进⾏数据处理；此时，每执⾏⼀次迭代，就对所 有的⽆状态的中间操作进⾏数据处理，⽽对有状态的中间操作，就需要迭代处理完所有的数据，再进⾏处理操作；最后就是进
⾏终结操作的数据处理。
在并⾏处理操作中，Stream对中间操作基本跟串⾏处理⽅式是⼀样的，但在终结操作中，Stream将结合ForkJoin框架对集合进⾏切⽚处理，ForkJoin框架将每个切⽚的处理结果Join合并起来。最后就是要注意Stream的使⽤场景。
思考题
这⾥有⼀个简单的并⾏处理案例，请你找出其中存在的问题。
//使⽤⼀个容器装载100个数字，通过Stream并⾏处理的⽅式将容器中为单数的数字转移到容器parallelList List for (int i = 0; i <100; i++) { integerList.add(i); } List .parallel() .filter(i->i%2==1) .forEach(i->parallelList.add(i)); |
|---|
期待在留⾔区看到你的答案。也欢迎你点击“请朋友读”，把今天的内容分享给身边的朋友，邀请他⼀起学习。
精选留⾔ <br />⼩⽩猪<br />思考题，由于流是并⾏处理，parallelList会存在并发问题，应该使⽤collect⽅法聚合<br />2019-06-01 10:01<br />(´⽥ω⽥`)<br />感觉这⼀节课已经值回了整个课程的票价，给⽼师点赞！
思考题：Stream并⾏执⾏，⽆法确认每个元素的处理顺序，最后parallelList中的数字是⽆序的
2019-06-01 02:11
作者回复
思考题中的问题是在并⾏操作arraylist时，需要考虑线程安全问题
2019-06-01 20:45
Loubobooo
parallelList集合⾥呈现的是⽆序的数字，是这样吗？
2019-06-01 11:16
作者回复
对的，可能会出现少数字、⽆序以及异常情况
2019-06-02 13:15
⼀路看⻛景
⽼师您好，在容器盛⾏的微服务环境下，以及⼤数据处理流⾏的潮流中，我觉得stream的应⽤空间多少有些尴尬呢，不知是不是我的理解有误。即：单核容器运⾏的环境下stream没了性能优势，⼤数据的处理⼜有⼤数据平台去完成使命，所以是不是意味着我们可以从stream得到的最⼤收益变成了流式编程和函数式编程带来的代码易读和易⽤性了呢？
2019-06-01 17:46
作者回复
是的，但未必所有公司都有构建⼤数据的能⼒，⽽且⼀些公司有⾃⼰的中间件团队，例如⽂章开始说到的分表分库的查询操作
，使⽤stream的并⾏操作就有优势了
2019-06-01 21:00
Darren
最终的结果⽆序，且可能结果都是不正确的，因为ArrayList是线程不安全的
2019-06-04 11:59
QQ怪
是不是该把思考题中的arraylist换成线程安全的copyOnwriteList就可以解决线程不安全问题?
2019-06-03 23:55
作者回复
对的，但copyOnwriteList更适合某⼀时间段统⼀新增，且新增时避免⼤量操作容器发⽣。⽐较适合在深夜更新⿊名单类似的业务。
2019-06-04 09:27
⼩辉辉
ArrayList是线程不安全的集合，⽽当前⼜⽤了并⾏流去处理，所以会出现有异常、少数据或者正常输出结果这三种情况。
2019-06-02 17:26
bro.
⽼师，这么早更新，读完感觉跟rxjava设计思想很接近，不订阅前⾯过滤条件都不会真正的运⾏！
2019-06-01 07:28
N
⽼师有个问题请教⼀下，公司业务代码中有⼤量stream对集合遍历，过滤，聚合的⽤法，但都是串⾏的，因为⼤部分数据量不是很⼤，请问数据量多⼤的时候才有必要使⽤并⾏提⾼效率呢？
2019-07-14 13:27
作者回复
上万数量级使⽤并⾏可以提⾼效率。
2019-07-14 15:40
阿厚
⽼师，请教2个问题：
1.有什么分表分库中间件推荐么？
2.分表分库以后，查询分⻚怎么办呢？
2019-06-06 09:09
作者回复
之前⽤过sharing-jdbc以及mycat，⼀个明显的区别是sharing-jdbc是嵌⼊⽅式，⽽mycat是基于proxy，所以理论上来说 proxy⽅式会有性能损耗。现在我们在使⽤sharing-jdbc，这⾥不打⼴告，两个中间件都有⾃⼰的优势。
分⻚查询是基于我这篇⽂章说的⽅式，将每个分表的数据结果集查询出来，通过归并排序计算出。具体的实现⽅式有区别，本次专栏的后⾯课程也会具体讲到。
2019-06-07 10:17
郑晨Cc
课程好值啊 全是⼲货
2019-06-03 09:57
Liam
parallel Stream 的并发机制是通过ForkJoinPool实现的，它的通⽤线程池是⼀个⽆界队列，想问下，数据量很⼤的时候，⽐如1
w个元素，它分⽚的依据是什么，每个分⽚多⼤；⼦任务⽐较多的时候，会不会严重消耗内存以及频繁触发GC等
2019-06-01 09:55
Liam
并发操作⼀个ArrayList，会有线程安全问题？
2019-06-01 09:36
作者回复
对的
2019-06-01 21:43
圣⻄罗
⽼师，现在⽹上有些说法做测试⽤lambda⽐普通for循环速度慢五倍，因此有⼈拒绝⽤。实际情况是什么样呢？如果我⾃⼰想测
，应该怎么尽可能排除外因⼲扰，测⼀下他们的实际效率对⽐？
2019-06-01 09:01
作者回复
当应⽤程序以前没有使⽤lambda表达式时，会动态⽣成lambda⽬标对象，这是导致慢的实际原因。我们可以在运⾏加载后，
也就是初次测试之后，紧接着后⾯加⼏个for循环，再测试⼏次，对⽐下性能。
虽然单独使⽤lambda表达式在初次运⾏时要⽐传统⽅式慢很多，但结合stream的并⾏操作，在多核环境下还有有优势的。
2019-06-02 12:26
陆离
思考题中这样的⽅式会造成null值和缺值
因为arraylist不是线程安全的，例如线程⼀在size++后准备给index为size+1的位置赋值，这个时候第⼆个线程⼜给size++，这个线程⼀赋值的index就变成了size+2,在线程⼀赋值后，线程⼆⼜在size+2的位置赋值。
这样的结果就是size+1的位置没有值null,size+2的位置为线程⼆赋的值，线程⼀赋的值被覆盖。 正确的⽅式应该是使⽤collect()
2019-06-01 08:12
师琳博
通过对⽐发现（在多核场景），能⽤stream并⾏就⽤，不能⽤就⽤常规，stream串⾏好像没有任何优势可⾔。是不是多有场景中都不建议使⽤stream串⾏呢？
2019-07-08 12:35
作者回复
是的，stream在多核机器下并⾏处理⼤数据量优势明显。
2019-07-08 16:24
⼀路奔跑
ArraryList是⾮线性安全的，并⾏流处理会出现越界或者重复或者少元素的情况！这个坑我踩过！
2019-07-08 08:12
作者回复
过来⼈，印象应该深刻
2019-07-08 16:31
ok绷
parallelList是⾮线程安全的，可以使⽤线程安全的集合类，但是不知道到使⽤stream的collect⽅法可以吗？
2019-07-02 08:24
作者回复
可以
2019-07-04 09:28
男朋友
让我想到了REDIS,虽然是单线程的,但是redis是等数据来了才处理,⽽不是⼀连接就处理或者等待的.
2019-07-01 17:55
⾸先是测试结果，⽆序的问题是其⼀，不过也不算问题。另⼀个问题是，有可能会最后结果只有47~49个值的现象（实际值应该为50个）。并且多次循环的话会报下标越界。
⾃认为是ArrayList的并发问题。
2019-06-10 13:10
作者回复
对了！
2019-06-10 22:02
若有收获，就点个赞吧
0 人点赞
