Stream API

  1. Java 8集合中的Stream相当于高级版的Iterator
    • Stream API通过Lambda表达式对集合进行各种非常便利高效的聚合操作,或者大批量数据操作
  2. Stream的聚合操作与数据库SQL的聚合操作sorted、filter、map等非常类似
  3. 在数据操作方面,Stream不仅可以通过串行的方式实现数据操作,还可以通过并行的方式处理大批量数据,提高处理效率
1
2
3
4
5
6
7
8
// java.util.Collection
default Stream<E> stream() {
return StreamSupport.stream(spliterator(), false);
}

default Stream<E> parallelStream() {
return StreamSupport.stream(spliterator(), true);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@Data
class Student {
private Integer height;
private String sex;
}

Map<String, List<Student>> map = Maps.newHashMap();
List<Student> list = Lists.newArrayList();

// 传统的迭代方式
for (Student student : list) {
if (student.getHeight() > 160) {
String sex = student.getSex();
if (!map.containsKey(sex)) {
map.put(sex, Lists.newArrayList());
}
map.get(sex).add(student);
}
}
// Stream API,串行实现
map = list.stream().filter((Student s) -> s.getHeight() > 160).collect(Collectors.groupingBy(Student::getSex));

// Stream API,并行实现
map = list.parallelStream().filter((Student s) -> s.getHeight() > 160).collect(Collectors.groupingBy(Student::getSex));

优化遍历

Stream操作分类

  1. 分为两大类:中间操作(Intermediate operations)和终结操作(Terminal operations)
  2. 中间操作只对操作进行了记录,即只会返回一个流,不会进行计算操作,而终结操作是实现了计算操作
  3. 中间操作又分为无状态(Stateless)操作和有状态(Stateful)操作
    • 无状态操作:元素的处理不受之前元素的影响
    • 有状态操作:该操作只有拿到所有元素之后才能继续下去
  4. 终结操作又分为短路(Short-circuiting)操作与非短路(UnShort-circuiting)操作
    • 短路操作:遇到某些符合条件的元素就可以得到最终结果
    • 非短路操作:必须处理完所有元素才能得到最终结果
  5. 通常会将中间操作称为懒操作,正是因为懒操作结合终结操作,数据源构成的处理管道(Pipeline),实现了Stream的高效

Stream源码实现

  1. BaseStream和Stream为最顶端的接口类
    • BaseStream定义了流的基本接口方法,如spliterator、isParallel等
    • Stream定义了流的常用操作方法,如map、filter等
  2. ReferencePipeline是一个结构类,通过定义内部类组装各种操作流
    • 内部定义了Head、StatelessOp和StatefulOp三个内部类,实现了BaseStream和Stream的接口方法
  3. Sink接口定义每个Stream操作之间关系的协议,包含了begin、end、cancellationRequested、accept方法
    • ReferencePipeline最终会将整个Stream流操作组装成一个调用链
    • 而调用链上的每个Stream操作的上下文关系就是通过Sink接口来定义实现的

Stream操作叠加

  1. 一个Stream的各个操作是由处理管道组装的,并统一完成数据处理
  2. 在JDK中,每次的中断操作都会以使用阶段(Stage)命名
  3. 管道结构通常是由ReferencePipeline类实现的,ReferencePipeline包含Head、StatelessOp、StatefulOp三个内部类
    • Head类主要用来定义数据源操作,初次调用.stream()时,会初次加载Head对象
    • 接着加载中间操作,分为StatelessOp对象和StatefulOp对象
      • 此时的Stage并没有执行,而是通过AbstractPipeline生成了中间操作的Stage链表
    • 当调用终结操作时,会生成一个最终的Stage
      • 通过这个Stage触发之前的中间操作,从最后一个Stage开始,递归产生一个Sink链

样例

1
2
3
4
5
6
List<String> names = Arrays.asList("张三", "李四", "王老五", "李三", "刘老四", "王小二", "张四", "张五六七");
String maxLenStartWithZ = names.stream()
.filter(name -> name.startsWith("张"))
.mapToInt(String::length)
.max()
.toString();

names是ArrayList集合,names.stream会调用集合类基础接口Collection的stream方法

1
2
3
default Stream<E> stream() {
return StreamSupport.stream(spliterator(), false);
}

Collection.stream方法会调用StreamSupport.stream方法,方法中初始化了一个ReferencePipeline的Head内部类对象

1
2
3
4
5
6
public static <T> Stream<T> stream(Spliterator<T> spliterator, boolean parallel) {
Objects.requireNonNull(spliterator);
return new ReferencePipeline.Head<>(spliterator,
StreamOpFlag.fromCharacteristics(spliterator),
parallel);
}

调用filter和map,两者都是无状态的中间操作,因此并没有执行任何操作,只是分别创建了一个Stage标识用户的每一次操作
通常情况下,Stream的操作需要一个回调函数,所以一个完整的Stage是由数据来源、操作、回调函数组成的三元组表示

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@Override
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);
}
};
}
};
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@Override
@SuppressWarnings("unchecked")
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));
}
};
}
};
}

new StatelessOp会调用父类AbstractPipeline的构造函数,该构造函数会将前后的Stage联系起来,生成一个Stage链表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
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类中

1
2
3
4
@Override
public final Optional<P_OUT> max(Comparator<? super P_OUT> comparator) {
return reduce(BinaryOperator.maxBy(comparator));
}

最后调用AbstractPipeline的wrapSink方法,生成一个Sink链表,Sink链表中的每一个Sink都封装了一个操作的具体实现

1
2
3
4
5
6
7
8
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;
}

当Sink链表生成完成后,Stream开始执行,通过Spliterator迭代集合,执行Sink链表中的具体操作

1
2
3
4
5
6
7
8
9
10
11
12
13
@Override
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);
}
}
  1. java 8中的forEachRemaining会迭代集合
  2. 每迭代一次,都会执行一次filter操作,通过后就会触发map操作,然后将结果放入到临时数组object中,再进行下一次迭代
  3. 完成中间操作后,最后触发终结操作max

Stream并行处理

1
2
3
4
5
6
7
List<String> names = Arrays.asList("张三", "李四", "王老五", "李三", "刘老四", "王小二", "张四", "张五六七");
String maxLenStartWithZ = names.stream()
.parallel()
.filter(name -> name.startsWith("张"))
.mapToInt(String::length)
.max()
.toString();

Stream的并行处理在执行终结操作之前,跟串行处理的实现是一样的,在调用终结方法之后,会调用TerminalOp.evaluateParallel

1
2
3
4
5
6
7
8
9
10
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()));
}
  1. 并行处理指的是Stream结合了ForkJoin框架,对Stream处理进行了分片,Spliterator.estimateSize会估算出分片的数据量
  2. 通过预估的数据量获取最小处理单元的阈值,如果当前分片大小大于最小处理单元的阈值,就继续切分集合
  3. 每个分片都将会生成一个Sink链表,当所有分片操作完成后,ForkJoin框架将会合并分片任何结果集

合理使用Stream

  1. 在循环迭代次数较少的情况下,常规的迭代方式性能反而更好
  2. 在单核CPU服务器配置环境中,也是常规迭代方式更有优势
  3. 大数据循环迭代中,如果服务器是多核CPU的情况,采用Stream的并行迭代优势明显

小结

  1. Stream将整个操作分解成了链式结构,不仅简化了遍历操作,还为实现并行计算奠定了基础
  2. Stream将遍历元素的操作和对元素的计算分为中间操作终结操作
    • 中间操作又根据元素之间状态有无干扰分为有状态操作无状态操作,实现了链式结构中的不同阶段
  3. 串行处理
    • Stream在执行中间操作时,并不会做实际的数据操作处理,而是将这些中间操作串联起来,最终由终结操作触发
    • 生成一个数据处理链表,通过Java 8的Spliterator迭代器进行数据处理
  4. 并行处理
    • 对中间操作的处理跟串行处理的方式是一样的,但在终结操作中,Stream将结合ForkJoin框架对集合进行切片处理

参考资料

Java性能调优实战