的设计和实现(一)的源码-虎哥说创业

大家对Java8应该不陌生。本文主要从源码层面讨论Java8的设计与实现。

看看下面的示例代码：

 static class A {
        @Getter
        private String a;
        @Getter
        private Integer b;
        public A(String a, Integer b) {
            this.a = a;
            this.b = b;
        }
    }
    public static void main(String[] args) {
        List 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);
    }

在上面的代码中，实际上有几个步骤：

..地图..

一步一步来，真正调用的是方法：

 default Stream stream() {
        return StreamSupport.stream(spliterator(), false);
    }

() 方法生成一个对象，这意味着它可以拆分。这个主要用于中间的并行操作。在上面的例子中，因为它被调用了，=false。

.最终结果是一个 .Head 对象：

 public static  Stream stream(Spliterator spliterator, boolean parallel) {
        Objects.requireNonNull(spliterator);
        return new ReferencePipeline.Head(spliterator,
                                            StreamOpFlag.fromCharacteristics(spliterator),
                                            parallel);
    }

Head 类派生自并代表 .

有了这个Head对象之后，对其调用.map其实就是调用基类的.map方法：

   public final  Stream map(Function mapper) {
        Objects.requireNonNull(mapper);
        return new StatelessOp(this, StreamShape.REFERENCE,
                                     StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
            @Override
            Sink
 opWrapSink(int flags, Sink sink) {
                return new Sink.ChainedReference
(sink) {

                    @Override
                    public void accept(P_OUT u) {
                        downstream.accept(mapper.apply(u));
                    }
                };
            }
        };
    }

返回的是一个，表示无状态的操作符。这个类也是的子类。你可以看到它的构造函数。第一个参数this表示Head对象作为一个对象，即它的上游。 . 方法先不说，后面再说。

然后调用 . 方法，它仍然会返回到 . 方法：

public final Stream filter(Predicate predicate) {
        Objects.requireNonNull(predicate);
        return new StatelessOp
(this, StreamShape.REFERENCE,
                                     StreamOpFlag.NOT_SIZED) {
            @Override
            Sink
 opWrapSink(int flags, Sink
 sink) {
                return new Sink.ChainedReference(sink) {
                    @Override
                    public void begin(long size) {
                        downstream.begin(-1);
                    }
                    @Override
                    public void accept(P_OUT u) {
                        if (predicate.test(u))
                            downstream.accept(u);
                    }
                };
            }
        };
    }

可以看到还是生成了一个对象，只是名字变了。

最后，打电话。并继续回到 . 方法：

  public final  R collect(Collector collector) {
        A container;
        if (isParallel()
                && (collector.characteristics().contains(Collector.Characteristics.CONCURRENT))
                && (!isOrdered() || collector.characteristics().contains(Collector.Characteristics.UNORDERED))) {
            container = collector.supplier().get();
            BiConsumer 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);
    }

在前面的步骤中，.map 和. 实际上只是创建对象，但它们是不同的。懂spark/flink的就知道，它其实就是一个/sink。调用时，它实际上会触发其中每一个的执行。这就是我们经常听到的惰性，所有的操作只有遇到才会执行。

前面提到了=false，所以上面的实际执行逻辑是：

A container = evaluate(ReduceOps.makeRef(collector));
return collector.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)
               ? (R) container
               : collector.finisher().apply(container);
    }

在进入方法之前，我们先看一下.()，它实际上是根据生成的实例包裹了一层，并返回了一个对象（实际上）。

    public static  TerminalOp
    makeRef(Collector collector) {
        Supplier supplier = Objects.requireNonNull(collector).supplier();
        BiConsumer accumulator = collector.accumulator();
        BinaryOperator combiner = collector.combiner();
        class ReducingSink extends Box
                implements AccumulatingSink {
            @Override
            public void begin(long size) {
                state = supplier.get();
            }
            @Override
            public void accept(T t) {
                accumulator.accept(state, t);
            }
            @Override
            public void combine(ReducingSink other) {
                state = combiner.apply(state, other.state);
            }
        }
        return new ReduceOp(StreamShape.REFERENCE) {
            @Override
            public ReducingSink makeSink() {
                return new ReducingSink();
            }
            @Override
            public int getOpFlags() {
                return collector.characteristics().contains(Collector.Characteristics.UNORDERED)

                       ? StreamOpFlag.NOT_ORDERED
                       : 0;
            }
        };
    }