Java8 Lambda实现源码解析
阿
王逸(卫乐)约 11 分钟
王逸(卫乐)约 11 分钟
前言
Java8的lambda应该大家都比较熟悉了，本文主要从源码层面探讨一下lambda的设计和实现。基础示例与解析
先看下面的示例代码：static class A {@Getterprivate String a;@Getterprivate Integer b;public A(String a, Integer b) {this.a = a;this.b = b;}}public static void main(String[] args) {List<Integer> ret = Lists.newArrayList(new A("a", 1), new A("b", 2), new A("c", 3)).stream().map(A::getB).filter(b -> b >= 2).collect(Collectors.toList());System.out.println(ret);}
- ArrayList.stream
- .map
- .filter
- .collect
default Stream<E> stream() {return StreamSupport.stream(spliterator(), false);}
public static <T> Stream<T> stream(Spliterator<T> spliterator, boolean parallel) {Objects.requireNonNull(spliterator);return new ReferencePipeline.Head<>(spliterator,StreamOpFlag.fromCharacteristics(spliterator),parallel);}
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) {@OverrideSink<P_OUT> opWrapSink(int flags, Sink<R> sink) {return new Sink.ChainedReference<P_OUT, R>(sink) {@Overridepublic void accept(P_OUT u) {downstream.accept(mapper.apply(u));}};}};}
可以看到，仍然生成的是一个StatelessOp对象，只是它的upstream变了而已。 最后调用StatelessOp.collect，继续回到ReferencePipeline.collect方法：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) {@OverrideSink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {@Overridepublic void begin(long size) {downstream.begin(-1);}@Overridepublic void accept(P_OUT u) {if (predicate.test(u))downstream.accept(u);}};}};}
在前面几步，.map, .filter方法其实都只是创建StatelessOp对象，但是到collect就不一样了，了解spark/flink的就知道，collect其实是个action/sink，调用了collect，就会真实地触发这个stream上各个operator的执行。这也就是我们经常听到的lazy execution，所有的操作，只有碰到action的算子才会开始执行。 之前讲到这个stream的parallel=false，所以上面的实际执行逻辑是：public final <R, A> R collect(Collector<? super P_OUT, A, R> collector) {A container;if (isParallel()&& (collector.characteristics().contains(Collector.Characteristics.CONCURRENT))&& (!isOrdered() || collector.characteristics().contains(Collector.Characteristics.UNORDERED))) {container = collector.supplier().get();BiConsumer<A, ? super P_OUT> accumulator = collector.accumulator();forEach(u -> accumulator.accept(container, u));}else {container = evaluate(ReduceOps.makeRef(collector));}return collector.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)? (R) container: collector.finisher().apply(container);}
A container = evaluate(ReduceOps.makeRef(collector));return collector.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)? (R) container: collector.finisher().apply(container);}
上面代码可以看到，基本也就是直接调用了collector的实现，稍微需要注意的是，ReducingSink从Box派生，Box的意思就是盒子，它里面有个state成员，表示一个计算的状态。ReducingSink就是通过这个state，进行combine, accumulate操作（实际就是一个List)。 回到evaluate方法，它实际调用了：public static <T, I> TerminalOp<T, I>makeRef(Collector<? super T, I, ?> collector) {Supplier<I> supplier = Objects.requireNonNull(collector).supplier();BiConsumer<I, ? super T> accumulator = collector.accumulator();BinaryOperator<I> combiner = collector.combiner();class ReducingSink extends Box<I>implements AccumulatingSink<T, I, ReducingSink> {@Overridepublic void begin(long size) {state = supplier.get();}@Overridepublic void accept(T t) {accumulator.accept(state, t);}@Overridepublic void combine(ReducingSink other) {state = combiner.apply(state, other.state);}}return new ReduceOp<T, I, ReducingSink>(StreamShape.REFERENCE) {@Overridepublic ReducingSink makeSink() {return new ReducingSink();}@Overridepublic int getOpFlags() {return collector.characteristics().contains(Collector.Characteristics.UNORDERED)? StreamOpFlag.NOT_ORDERED: 0;}};}
terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));public <P_IN> R evaluateSequential(PipelineHelper<T> helper,Spliterator<P_IN> spliterator) {return helper.wrapAndCopyInto(makeSink(), spliterator).get();}
wrapSink代码：copyInto(wrapSink(Objects.requireNonNull(sink)), spliterator);return sink;
可以看到，这里就是将pipeline从后至前，分别调用每个pipeline的opWrapSink方法，就是一个责任链的模式。opWrapSink可以看上面map的opWrapSink的filter的opWrapSink实现，map的很简单，直接调用mapper.apply，实际上就是A::getB方法，filter的也很简单，调用的是 predicate.test 方法。 接下来到copyInto方法，到这里才会有真正的执行逻辑：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;}
它会走入到这部分的逻辑中：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);}}
wrappedSink.begin(spliterator.getExactSizeIfKnown());spliterator.forEachRemaining(wrappedSink);wrappedSink.end();
这里的action参数，就是上面经过责任链封装的Sink（它也是Consumer的子类）。public void forEachRemaining(Consumer<? super E> action) {// ...if ((i = index) >= 0 && (index = hi) <= a.length) {for (; i < hi; ++i) {@SuppressWarnings("unchecked") E e = (E) a[i];action.accept(e);}if (lst.modCount == mc)return;}// ...
而这里调用action.accept，就会通过责任链来一层层调用每个算子的accept，我们从map的accept开始：
@OverrideSink<P_OUT> opWrapSink(int flags, Sink<R> sink) {return new Sink.ChainedReference<P_OUT, R>(sink) {@Overridepublic void accept(P_OUT u) {downstream.accept(mapper.apply(u));}};}
直接继承此接口的，是如IntStream, LongStream，DoubleStream等，这些是在BaseStream基础上，提供了filter, map, mapToObj, distinct等算子的接口，但是这些算子，是限定类型的，如IntStream.filter, 它接受的就是 IntPredicate， 而不是常规的Predicate；map方法也是，接受的是 IntUnaryOperator。 IntStream, LongStream这些都是接口，也就是仅仅用来描述算子的。它们的实现都是基于Pipeline的，基类为 AbstractPipeline，它的几个关键成员变量：public interface BaseStream<T, S extends BaseStream<T, S>>extends AutoCloseable {/*** 返回stream中元素的迭代器*/Iterator<T> iterator();/*** 返回stream中元素的spliterator，用于并行执行*/Spliterator<T> spliterator();/*** 是否并行*/boolean isParallel();/*** 返回串行的stream，即强制parallel=false*/S sequential();/*** 返回并行的stream，即强制parallel=true*/S parallel();// ...}
/*** 最顶上的pipeline，即Head*/private final AbstractPipeline sourceStage;/*** 直接上游pipeline*/private final AbstractPipeline previousStage;/*** 直接下游pipeline*/@SuppressWarnings("rawtypes")private AbstractPipeline nextStage;/*** pipeline深度*/private int depth;/*** head的spliterator*/private Spliterator<?> sourceSpliterator;// ...
双流concat的场景示例及解析
接下来看一个相对比较复杂的例子，双流concat的场景，代码如下：上面代码中，先分别创建两个IntStream：s1, s2。然后进行concat操作，生成s2，最后调用sum操作做reduce。 代码分析还是从sink开始，reduce跟前面的collect类似，实际会基于s3这个stream, 在AbstractPipeline.evaluate方法中执行：static class Mapper1 implements IntUnaryOperator {@Overridepublic int applyAsInt(int operand) {return operand * operand;}}static class Filter1 implements IntPredicate {@Overridepublic boolean test(int value) {return value >= 2;}}static class Mapper2 implements IntUnaryOperator {@Overridepublic int applyAsInt(int operand) {return operand + operand;}}static class Filter2 implements IntPredicate {@Overridepublic boolean test(int value) {return value >= 10;}}static class Mapper3 implements IntUnaryOperator {@Overridepublic int applyAsInt(int operand) {return operand * operand;}}static class Filter3 implements IntPredicate {@Overridepublic boolean test(int value) {return value >= 10;}}public static void main(String[] args) {IntStream s1 = Arrays.stream(new int[] {1, 2, 3}).map(new Mapper1()).filter(new Filter1());IntStream s2 = Arrays.stream(new int[] {4, 5, 6}).map(new Mapper2()).filter(new Filter2());IntStream s3 = IntStream.concat(s1, s2).map(new Mapper3()).filter(new Filter3());int sum = s3.sum();}
terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));final <P_IN, S extends Sink<E_OUT>> S wrapAndCopyInto(S sink, Spliterator<P_IN> spliterator) {copyInto(wrapSink(Objects.requireNonNull(sink)), spliterator);return sink;}
Head(concated s1 + s2 stream) -> Mapper3 -> Filter3 -> ReduceOp(sum) 到目前为止，我们还只看到s3的逻辑，那么s1和s2两个stream的mapper和filter逻辑在哪里呢，接着看下面的copyInto方法：
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();// ...
public void forEachRemaining(Consumer<? super T> consumer) {if (beforeSplit)aSpliterator.forEachRemaining(consumer);bSpliterator.forEachRemaining(consumer);}
所以就走到了 IntWrappingSpliterator.foreachMaining方法中：// 包装的原始pipelinefinal PipelineHelper<P_OUT> ph;// 原始pipeline的spliteratorSpliterator<P_IN> spliterator;
public void forEachRemaining(IntConsumer consumer) {if (buffer == null && !finished) {Objects.requireNonNull(consumer);init();ph.wrapAndCopyInto((Sink.OfInt) consumer::accept, spliterator);finished = true;}// ...
AbstractPipeline.wrapAndCopyInto -> AbstractPipeline.wrapSink -> AbstractPipeline.copyInto
方法中，而在这时的wrapSink中，现在的pipeline就是s1/s2了，这时就会对s1/s2下面的所有算子，调用AbstractPipeline.opWrapSink串联起来，以s1为例就变成： Head(array[1,2,3]) -> Mapper1 -> Filter1 -> Mapper3 -> Filter3 -> ReduceOp(sum) 这样s1流跟s3流就串起来执行完成了，然后就是s2和s3流串起来执行。阿里云开发者社区&阿里开发者微信公众号联合推出的创作挑战赛进入征稿倒计时！活动截止11月30日，我们邀请你，前往社区分享你的技术思考，让创作创造价值！
点击阅读原文查看详情。