摘要：正常情況下，一個(gè)流在執(zhí)行一次終端操作之后便結(jié)束了。本文通過復(fù)制流內(nèi)數(shù)據(jù)的方式，曲折的實(shí)現(xiàn)了同一個(gè)流上執(zhí)行多次操作。只是思路，其性能并不一定高效，尤其是數(shù)據(jù)都在內(nèi)存中處理時(shí)復(fù)制的開銷很大。但如果流涉及大量，也許性能會有提高。
正常情況下，一個(gè)流在執(zhí)行一次終端操作之后便結(jié)束了。本文通過復(fù)制流內(nèi)數(shù)據(jù)的方式，曲折的實(shí)現(xiàn)了同一個(gè)流上執(zhí)行多次操作。
Demo只是思路，其性能并不一定高效，尤其是數(shù)據(jù)都在內(nèi)存中處理時(shí)復(fù)制的開銷很大。但如果流涉及大量I／O，也許性能會有提高。
public class StreamForker {
private final Stream stream;
private final Map, ?>> forks = new HashMap<>();
public StreamForker(Stream stream) {
this.stream = stream;
}
public StreamForker fork(Object key, Function, ?> f) {
forks.put(key, f);
return this;
}
public Results getResults() {
ForkingStreamConsumer consumer = build();
try {
stream.sequential().forEach(consumer);
} finally {
consumer.finish();
}
return consumer;
}
private ForkingStreamConsumer build() {
List> queues = new ArrayList<>();
Map> actions = forks.entrySet().stream().reduce(new HashMap>(),
(map, e) -> {
map.put(e.getKey(), getOperationResult(queues, e.getValue()));
return map;
}, (m1, m2) -> {
m1.putAll(m2);
return m1;
});
return new ForkingStreamConsumer<>(queues, actions);
}
private Future getOperationResult(List> queues, Function, ?> f) {
BlockingQueue queue = new LinkedBlockingQueue<>();
queues.add(queue);
Spliterator spliterator = new BlockingQueueSpliterator<>(queue);
Stream source = StreamSupport.stream(spliterator, false);
return CompletableFuture.supplyAsync(() -> f.apply(source));
}
}
accept方法將原始流中所有的數(shù)據(jù)添加到各個(gè)BlockingQueue內(nèi)，此處實(shí)現(xiàn)了復(fù)制
class ForkingStreamConsumer implements Consumer, Results {
static final Object END_OF_STREAM = new Object();
private final List> queues;
private final Map> actions;
public ForkingStreamConsumer(List> queues, Map> actions) {
this.queues = queues;
this.actions = actions;
}
@Override
public void accept(T t) {
queues.forEach(q -> q.add(t));
}
@SuppressWarnings("unchecked")
void finish() {
accept((T) END_OF_STREAM);
}
@SuppressWarnings("unchecked")
@Override
public R get(Object key) {
try {
return ((Future) actions.get(key)).get();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
此處重寫了tryAdvance接口，只是簡單的從BlockingQueue中取出數(shù)據(jù)，執(zhí)行action。業(yè)務(wù)邏輯中復(fù)制流是為了做什么事情，action就是這件事情。ForkingStreamConsumer.END_OF_STREAM是Queue中數(shù)據(jù)結(jié)束的標(biāo)示
class BlockingQueueSpliterator implements Spliterator {
private final BlockingQueue q;
BlockingQueueSpliterator(BlockingQueue q) {
this.q = q;
}
@Override
public boolean tryAdvance(Consumer action) {
T t;
while (true) {
try {
t = q.take();
break;
} catch (InterruptedException e) {
}
}
if (t != ForkingStreamConsumer.END_OF_STREAM) {
action.accept(t);
return true;
}
return false;
}
@Override
public Spliterator trySplit() {
return null;
}
@Override
public long estimateSize() {
return 0;
}
@Override
public int characteristics() {
return 0;
}
}
文章版權(quán)歸作者所有，未經(jīng)允許請勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。
轉(zhuǎn)載請注明本文地址：http://systransis.cn/yun/77289.html
