Java8 Stream
Stream講解示例: 找出前3短的單詞并轉(zhuǎn)成大寫形式
List<String> words = Arrays.asList(
"One",
"Two",
"Three",
"Four",
"Five",
"Six"
);
實現(xiàn)一：命令式——迭代：
@Test
public void test_takeTop3ShortWords_iterative() {
List<String> filtered = new ArrayList<>();
for(String word : words){
if(word.length()<=4){
filtered.add(word);
}
}
filtered.sort(Comparator.comparing(String::length));
List<String> top3 = filtered.subList(0, Math.min(3, filtered.size())); //節(jié)省了一個迭代
List<String> result = new ArrayList<>(top3);
for(String word : top3){
result.add(word.toUpperCase());
}
}
缺點：
- 代碼過于繁雜, 不太好直接看出意圖炕桨。
- 代碼的中間變量過多，命名困難穗慕。
實現(xiàn)二：聲明式——stream
@Test
public void test_takeTop3ShortWords_stream(){
List<String> result = words.stream()
.filter(s-> s.length()<=4)
.sorted(Comparator.comparing(String::length))
.map(String::toUpperCase)
.limit(3)
.collect(Collectors.toList());
}
Stream的實現(xiàn)原理
操作的分類
操作的分類
操作的串聯(lián)
- 通過相關(guān)的操作對象PipelineHelper的構(gòu)造函數(shù)，將操作先后關(guān)系維護起來（是反向的），維護一個代碼塊Sink鏈?zhǔn)浇Y(jié)構(gòu): D(C(B(A)))
- 調(diào)用wrapSink,將代碼塊正向串連起來: A->B->C->D
- 分別調(diào)用begin、foreach accecpt, end 代碼塊
操作的協(xié)作契約
interface Sink<T> extends Consumer<T> {
void begin(long size); //執(zhí)行一些初始化工作眠寿，如排序時要建一個臨時表用于存儲排序后的數(shù)據(jù)
void accept(T t); //繼承自Consumer，指明如何處理單個數(shù)據(jù)红选。
void end(); //指明本操作結(jié)束時（所有數(shù)據(jù)都已經(jīng)處理完）澜公，如何進行后續(xù)的處理姆另。
boolean cancellationRequested(); //當(dāng)下游不再需要更多數(shù)據(jù)時喇肋，可以用這個通知到上游操作坟乾。一般為短路操作。
}
自帶拆分功能的新版迭代器-數(shù)據(jù)流的驅(qū)動器
public interface Iterator<E> {
boolean hasNext(); //是否還有更多數(shù)據(jù)
E next(); //獲取下一個數(shù)據(jù)
void remove(); //刪除最近一個數(shù)據(jù)
default void forEachRemaining(Consumer<? super E> action) {
Objects.requireNonNull(action);
while (hasNext())
action.accept(next());
}
}
public interface Spliterator<T> {
long estimateSize(); //估算還有多少數(shù)據(jù)待迭代
boolean tryAdvance(Consumer<? super T> action); //處理單個數(shù)據(jù)
Spliterator<T> trySplit(); //用于并行流分解子任務(wù)
default void forEachRemaining(Consumer<? super T> action) {//默認(rèn)調(diào)用tryAdvance去遍歷蝶防，類似foreach
do { } while (tryAdvance(action));
}
}
流程示例
流程示例
Collection.stream()
- StreamSupport.stream(spliterator(), false)
- Spliterators.spliterator(this, 0)
- new ReferencePipeline.Head
AbstractPipeline
AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) {
if (previousStage.linkedOrConsumed)
throw new IllegalStateException(MSG_STREAM_LINKED);
previousStage.linkedOrConsumed = true;
previousStage.nextStage = this;
this.previousStage = 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;
}
@Override
final <P_IN, S extends Sink<E_OUT>> S wrapAndCopyInto(S sink, Spliterator<P_IN> spliterator) {
copyInto(wrapSink(Objects.requireNonNull(sink)), spliterator);
return sink;
}
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;
}
@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);
}
}
Spliterator
class ArraySpliterator<E> implements Spliterator<E> {
@SuppressWarnings("unchecked")
@Override
public void forEachRemaining(Consumer<? super T> action) {
Object[] a; int i, hi; // hoist accesses and checks from loop
if (action == null)
throw new NullPointerException();
if ((a = array).length >= (hi = fence) &&
(i = index) >= 0 && i < (index = hi)) {
do { action.accept((T)a[i]); } while (++i < hi);
}
}
}
filter
@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);
}
};
}
};
}
sorted
- SortedOps.makeRef(this)
class OfRef<T> extends ReferencePipeline.StatefulOp<T, T> {
@Override
public Sink<T> opWrapSink(int flags, Sink<T> sink) {
Objects.requireNonNull(sink);
// If the input is already naturally sorted and this operation
// also naturally sorted then this is a no-op
if (StreamOpFlag.SORTED.isKnown(flags) && isNaturalSort)
return sink;
else if (StreamOpFlag.SIZED.isKnown(flags))
return new SizedRefSortingSink<>(sink, comparator);
else
return new RefSortingSink<>(sink, comparator);
}
}
class RefSortingSink<T> extends AbstractRefSortingSink<T> {
private ArrayList<T> list;
RefSortingSink(Sink<? super T> sink, Comparator<? super T> comparator) {
super(sink, comparator);
}
@Override
public 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>();
}
@Override
public void end() {
list.sort(comparator);
downstream.begin(list.size());
if (!cancellationWasRequested) {
list.forEach(downstream::accept);
}
else {
for (T t : list) {
if (downstream.cancellationRequested()) break;
downstream.accept(t);
}
}
downstream.end();
list = null;
}
@Override
public void accept(T t) {
list.add(t);
}
}
map
@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));
}
};
}
};
}
limit
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;
@Override
public void begin(long size) {
downstream.begin(calcSize(size, skip, m));
}
@Override
public void accept(T t) {
if (n == 0) {
if (m > 0) {
m--;
downstream.accept(t);
}
}
else {
n--;
}
}
@Override
public boolean cancellationRequested() {
return m == 0 || downstream.cancellationRequested();
}
};
}
collect
ReduceOp
@Override
public <P_IN> R evaluateSequential(PipelineHelper<T> helper, Spliterator<P_IN> spliterator) {
return helper.wrapAndCopyInto(makeSink(), spliterator).get(); //new ReducingSink()
}
class ReducingSink extends Box<R> implements AccumulatingSink<T, R, ReducingSink> {
@Override
public void begin(long size) {
state = seedFactory.get();
}
@Override
public void accept(T t) {
accumulator.accept(state, t);
}
@Override
public void combine(ReducingSink other) {
reducer.accept(state, other.state);
}
}
