Java 8之后我们对Stream的使用都已经习以为常了,它帮助我们从怎么做的细节里脱身,只要告诉它做什么即可。这一篇文章我们主要讲Java Stream的实现原理,手写一个Stream框架,然后再来讲解Java Stream的核心类,做到知其然知其所以然。
我们先看一个Stream简单用例,输入一堆字符串创建流(of),取前5个(limit)、操作每个字符串取首字母(map),最后用终结操作符收集所有的首字母到List中(collect)。
List<Character> data = Stream.of("a", "b", "c", "d", "e", "f", "g").limit(5).map(x -> x.charAt(0)).collect(Collectors.toList());
System.out.println(data);
1. 创建Stream
假设我们要手写一个Stream类,你会怎么实现呢? 要实现Stream.of其实很简单,只要将of提供的参数保存在一个类实例中即可,我们定义了一个ArrayStream来实现这个动作
public static interface MyStream<T> {static <T> MyStream<T> of(T... ts) {return new OfStream<>(ts);}
}public static class ArrayStream<T> implements MyStream<T> {private int fromIdx;private int toIdx;private T[] data;public ArrayStream(T... ts) {this.data = ts;this.fromIdx = 0;this.toIdx = ts.length;}}
2. 支持limit
在ArrayStream的基础上支持limit操作就很容了,只要调整fromIdx、toIdx的值就即可。不过这里我们通过新定义一个类SliceStream来实现,在limit方法中重新创建SliceStream对象。
public static class SliceStream<T> implements MyStream<T> {private T[] data;private int fromIdx;private int toIdx;public SliceStream(T[] data, int fromIdx, int toIdx) {this.data = data;this.fromIdx = fromIdx;this.toIdx = toIdx;}@Overridepublic MyStream<T> limit(int n) {return new SliceStream<T>(this.data, fromIdx, Math.min(fromIdx + n, toIdx));}
}
在ArrayStream的limit也要使用SliceStream的实例。有的同学可能会疑惑为什么不直接用ArrayStream,修改fromIdx/toIdx,这个案例是可以的,新增类只是为了模拟JDK实现的策略。
public static class ArrayStream<T> implements MyStream<T> {...@Overridepublic MyStream<T> limit(int n) {return new SliceStream<>(this.data, 0, Math.min(data.length, n));}...
}
3. 支持map
照着limit的实现如法炮制,我们定义一个MappedStream类,持有转换函数(Funtion),持有下游MyStream,后续读取MyStream时,先用转换函数操作数据。我们来看下代码
public static class MappedStream<T, R> implements MyStream<R> {private Function<T, R> apply;private MyStream<T> stream;public MappedStream(MyStream<T> stream, Function<T, R> apply) {this.stream = stream;this.apply = apply;}...@Overridepublic void forEach(Consumer<R> consumer) {stream.forEach((T t) -> {R r = apply.apply(t);consumer.accept(r);});}
}
在ArrayStream的map方法中,我们讲ArrayStream包装到MappedStream后再返回
public static class ArrayStream<T> implements MyStream<T> {...@Overridepublic <R> MyStream<R> map(Function<T, R> map) {return new MappedStream<>(this, map);}...
}
4. 支持collect
Java Stream讲这类操作称为终结操作,它是实际动作发生的位置。我们定义一个Collector接口,supplier生成中间结果对象,accumulate应用每个元素到中间结果,finisher转换为最终结果
public static interface Collector<T, A, R> {public Supplier<A> supplier();public void accumulate(BiConsumer<T, R> consumer);Function<A, R> finisher();
}
下面是Collector的一个实现,在Collector接口提供一个工厂方法(toList),方便将stream转为list
public static class CollectorImpl<T, A, R> implements Collector<T, A, R> {private Supplier<A> supplier;private BiConsumer<T, A> accumulate;private Function<A, R> finisher;public CollectorImpl(Supplier<A> supplier, BiConsumer<T, A> accumulate, Function<A, R> finisher) {this.supplier = supplier;this.accumulate = accumulate;this.finisher = finisher;}@Overridepublic Supplier<A> supplier() {return supplier;}@Overridepublic BiConsumer<T, A> accumulate() {return accumulate;}@Overridepublic Function<A, R> finisher() {return finisher;}
}public static interface Collector<T, A, R> {static <T> CollectorImpl<T, List<T>, List<T>> toList() {Supplier<List<T>> supplier = ArrayList<T>::new;BiConsumer<T, List<T>> accumulate = (T t, List<T> ls) -> ls.add(t);Function<List<T>, List<T>> finisher = Function.identity();return new CollectorImpl<>(supplier, accumulate, finisher);}
}
在ArrayStream里支持通过collect方法终结操作
public static class ArrayStream<T> implements MyStream<T> {...@Overridepublic <A, W> W collect(Collector<T, A, W> collector) {Supplier<A> supplier = collector.supplier();BiConsumer<T, A> consumer = collector.accumulate();Function<A, W> finisher = collector.finisher();A middle = supplier.get();for (int i = fromIdx; i < toIdx; i++) {T item = data[i];consumer.accept(item, middle);}return finisher.apply(middle);}...
}
测试代码如下,查看输出可以确定我们的实现是正常运作的
public static void main(String[] args) {List<Character> data = MyStream.of("a", "b", "c", "d", "e", "f", "g").limit(5).map(x -> x.charAt(0)).collect(Collector.toList());System.out.println(data);
}
这个手写stream框架的类图是这个样子的,主要涉及了6个类
5. JDK内置实现
上面我们手写了简单的Stream框架,目的是了解它的工作原理。JDK内置大体思路跟这个类似,但要复杂的多。内置实现中也提供了Stream接口,以及Stream接口的各种实现类,本质类似于我们手写框架类的ArrayStream、MappedStream,通过装饰器模式提供额外功能。下图是核心的Stream实现类
我们用一个最简单的案例来看看Java的执行流程
Stream.of(1, 2, 3).filter(i -> i > 2).findAny().ifPresent(System.out::println);
这个调用的时序图如下图,核心步骤包括:
- Stream.of调用后,通过StreamSupport.stream创建一个ReferencePipeline.Head类实例
- filter(Prediccate)过滤流的时候,会把Head类实例封装为StatelessOp匿名内部类实例,并覆写了onWrapSink方法,在数据提交给Sink之前会先经过Predicate过滤
- findAny,创建一个FindOps实例,StatelessOp对象会调用FindOps的evaluateSequentail方法
- FindOps的evaluateSequentail方法,调用PipelineHelper类(ReferencePipeline实现该接口)的copyInto方法,copyInto方法内部会检测当前操作是否支持短路
- 支持短路操作,调用Spliterator的tryAdvance方法,Sink被设置为cancellationRequested的话就不操作之后的元素
- 不支持短路操作,调用Spliterator.forEachRemaining方法
现在我们完全理解Java Stream的运行流程了,再回过头来看看Java Stream中核心概念,Java Stream将操作分为两类,一类是中间操作,一类是终结操作
- 中间操作,继续返回stream够后续处理
- 终结操作,返回结果或Optional结束计算
中间操作被分为有状态操作(StatefullOps类)、无状态操作(StatelessOps类),终结操作都以TerminateOp表示,是否短路是通过combinedFlags标记的。Stream上不同API对应不同的操作符。
6. 如何评价Java Stream
有不少的文章/数据于对Java Stream做过性能测试,认为它是低效的。公平的将从实现上来说,Java Stream并没有明显存在严重影响性能的涉及。只是因为它有复杂的调用结构(各种包装器)所以对性能略有影响,而多数测试使用的都是极其简单的CPU密集型计算,所以显得对性能影响很大(有的甚至说比正常迭代慢2倍),如果是正常的业务逻辑完全不需要担心Stream会导致问题,而它带来的可读性是实实在在的。
下一篇我们来说说Java Stream的API支持哪些能力,怎么使用,如何扩展,尽情期待。