無狀態

unordered()

如果本身是無序的，則返回this，不做任何處理。

public Stream<P_OUT> unordered() {if (!isOrdered())return this;return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE, StreamOpFlag.NOT_ORDERED) {@OverrideSink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {return sink;}};}

filter()

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()

把mapper計算的結果傳給下游stream。

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));}};}};}

mapToInt()

public final IntStream mapToInt(ToIntFunction<? super P_OUT> mapper) {Objects.requireNonNull(mapper);return new IntPipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {@OverrideSink<P_OUT> opWrapSink(int flags, Sink<Integer> sink) {return new Sink.ChainedReference<P_OUT, Integer>(sink) {@Overridepublic void accept(P_OUT u) {downstream.accept(mapper.applyAsInt(u));}};}};}

mapToLong()

public final LongStream mapToLong(ToLongFunction<? super P_OUT> mapper) {Objects.requireNonNull(mapper);return new LongPipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {@OverrideSink<P_OUT> opWrapSink(int flags, Sink<Long> sink) {return new Sink.ChainedReference<P_OUT, Long>(sink) {@Overridepublic void accept(P_OUT u) {downstream.accept(mapper.applyAsLong(u));}};}};}

mapToDouble()

public final DoubleStream mapToDouble(ToDoubleFunction<? super P_OUT> mapper) {Objects.requireNonNull(mapper);return new DoublePipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {@OverrideSink<P_OUT> opWrapSink(int flags, Sink<Double> sink) {return new Sink.ChainedReference<P_OUT, Double>(sink) {@Overridepublic void accept(P_OUT u) {downstream.accept(mapper.applyAsDouble(u));}};}};}

flatMap()

mapper計算的結果是一個stream，對此stream調用forEach。傳入的函數是下游stream(Consumer接口）。

public final <R> Stream<R> flatMap(Function<? super P_OUT, ? extends Stream<? extends R>> mapper) {Objects.requireNonNull(mapper);// We can do better than this, by polling cancellationRequested when stream is infinitereturn new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {@OverrideSink<P_OUT> opWrapSink(int flags, Sink<R> sink) {return new Sink.ChainedReference<P_OUT, R>(sink) {@Overridepublic void begin(long size) {downstream.begin(-1);}@Overridepublic void accept(P_OUT u) {try (Stream<? extends R> result = mapper.apply(u)) {// We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach itif (result != null)result.sequential().forEach(downstream);}}};}};}

flatMapToInt()

flatMapToLong()

flatMapToDouble()

peek()

對每個元素應用Consumer。但是元素不做任何處理傳給下游stream。

public final Stream<P_OUT> peek(Consumer<? super P_OUT> action) {Objects.requireNonNull(action);return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,0) {@OverrideSink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {@Overridepublic void accept(P_OUT u) {action.accept(u);downstream.accept(u);}};}};}

有狀態
distinct()

static <T> ReferencePipeline<T, T> makeRef(AbstractPipeline<?, T, ?> upstream) {return new ReferencePipeline.StatefulOp<T, T>(upstream, StreamShape.REFERENCE,StreamOpFlag.IS_DISTINCT | StreamOpFlag.NOT_SIZED) {<P_IN> Node<T> reduce(PipelineHelper<T> helper, Spliterator<P_IN> spliterator) {// If the stream is SORTED then it should also be ORDERED so the following will also// preserve the sort orderTerminalOp<T, LinkedHashSet<T>> reduceOp= ReduceOps.<T, LinkedHashSet<T>>makeRef(LinkedHashSet::new, LinkedHashSet::add,LinkedHashSet::addAll);return Nodes.node(reduceOp.evaluateParallel(helper, spliterator));}@Override<P_IN> Node<T> opEvaluateParallel(PipelineHelper<T> helper,Spliterator<P_IN> spliterator,IntFunction<T[]> generator) {if (StreamOpFlag.DISTINCT.isKnown(helper.getStreamAndOpFlags())) {// No-opreturn helper.evaluate(spliterator, false, generator);}else if (StreamOpFlag.ORDERED.isKnown(helper.getStreamAndOpFlags())) {return reduce(helper, spliterator);}else {// Holder of null state since ConcurrentHashMap does not support null valuesAtomicBoolean seenNull = new AtomicBoolean(false);ConcurrentHashMap<T, Boolean> map = new ConcurrentHashMap<>();TerminalOp<T, Void> forEachOp = ForEachOps.makeRef(t -> {if (t == null)seenNull.set(true);elsemap.putIfAbsent(t, Boolean.TRUE);}, false);forEachOp.evaluateParallel(helper, spliterator);// If null has been seen then copy the key set into a HashSet that supports null values// and add nullSet<T> keys = map.keySet();if (seenNull.get()) {// TODO Implement a more efficient set-union view, rather than copyingkeys = new HashSet<>(keys);keys.add(null);}return Nodes.node(keys);}}@Override<P_IN> Spliterator<T> opEvaluateParallelLazy(PipelineHelper<T> helper, Spliterator<P_IN> spliterator) {if (StreamOpFlag.DISTINCT.isKnown(helper.getStreamAndOpFlags())) {// No-opreturn helper.wrapSpliterator(spliterator);}else if (StreamOpFlag.ORDERED.isKnown(helper.getStreamAndOpFlags())) {// Not lazy, barrier required to preserve orderreturn reduce(helper, spliterator).spliterator();}else {// Lazyreturn new StreamSpliterators.DistinctSpliterator<>(helper.wrapSpliterator(spliterator));}}@OverrideSink<T> opWrapSink(int flags, Sink<T> sink) {Objects.requireNonNull(sink);if (StreamOpFlag.DISTINCT.isKnown(flags)) {return sink;} else if (StreamOpFlag.SORTED.isKnown(flags)) {return new Sink.ChainedReference<T, T>(sink) {boolean seenNull;T lastSeen;@Overridepublic void begin(long size) {seenNull = false;lastSeen = null;downstream.begin(-1);}@Overridepublic void end() {seenNull = false;lastSeen = null;downstream.end();}@Overridepublic void accept(T t) {if (t == null) {if (!seenNull) {seenNull = true;downstream.accept(lastSeen = null);}} else if (lastSeen == null || !t.equals(lastSeen)) {downstream.accept(lastSeen = t);}}};} else {return new Sink.ChainedReference<T, T>(sink) {Set<T> seen;@Overridepublic void begin(long size) {seen = new HashSet<>();downstream.begin(-1);}@Overridepublic void end() {seen = null;downstream.end();}@Overridepublic void accept(T t) {if (!seen.contains(t)) {seen.add(t);downstream.accept(t);}}};}}};} }

sorted()?

所有元素接收完之后，通過comparator排序。

調用forEach，參數為下游stream。

private static final class RefSortingSink<T> extends Sink.ChainedReference<T, T> {private final Comparator<? super T> comparator;private ArrayList<T> list;RefSortingSink(Sink<? super T> sink, Comparator<? super T> comparator) {super(sink);this.comparator = comparator;}@Overridepublic void begin(long size) {if (size >= Nodes.MAX_ARRAY_SIZE)throw new IllegalArgumentException(Nodes.BAD_SIZE);list = (size >= 0) ? new ArrayList<T>((int) size) : new ArrayList<T>();}@Overridepublic void end() {list.sort(comparator);downstream.begin(list.size());list.forEach(downstream::accept);downstream.end();list = null;}@Overridepublic void accept(T t) {list.add(t);}}

limit()

計算器減1，直到為0,不再傳給下游。

Sink<T> opWrapSink(int flags, Sink<T> sink) {return new Sink.ChainedReference<T, T>(sink) {long n = skip;long m = limit >= 0 ? limit : Long.MAX_VALUE;@Overridepublic void begin(long size) {downstream.begin(calcSize(size, skip, m));}@Overridepublic void accept(T t) {if (n == 0) {if (m > 0) {m--;downstream.accept(t);}}else {n--;}}@Overridepublic boolean cancellationRequested() {return m == 0 || downstream.cancellationRequested();}};}

skip()

同limit（）

總結

以上是生活随笔為你收集整理的Java8函数式编程（5）--中间操作（源码）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。