摘要：正常情況下，一個流在執行一次終端操作之后便結束了。本文通過復制流內數據的方式，曲折的實現了同一個流上執行多次操作。只是思路，其性能并不一定高效，尤其是數據都在內存中處理時復制的開銷很大。但如果流涉及大量，也許性能會有提高。
正常情況下，一個流在執行一次終端操作之后便結束了。本文通過復制流內數據的方式，曲折的實現了同一個流上執行多次操作。
Demo只是思路，其性能并不一定高效，尤其是數據都在內存中處理時復制的開銷很大。但如果流涉及大量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方法將原始流中所有的數據添加到各個BlockingQueue內，此處實現了復制
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中取出數據，執行action。業務邏輯中復制流是為了做什么事情，action就是這件事情。ForkingStreamConsumer.END_OF_STREAM是Queue中數據結束的標示
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;
}
}
文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。
轉載請注明本文地址：http://m.specialneedsforspecialkids.com/yun/77289.html
