Java8流的復(fù)制

smartlion 發(fā)布于2019-08-15 13:25 / 1888人閱讀

摘要：只是思路，其性能并不一定高效，尤其是數(shù)據(jù)都在內(nèi)存中處理時(shí)復(fù)制的開(kāi)銷(xiāo)很大。方法將原始流中所有的數(shù)據(jù)添加到各個(gè)內(nèi)，此處實(shí)現(xiàn)了復(fù)制此處重寫(xiě)了接口，只是簡(jiǎn)單的從中取出數(shù)據(jù)，執(zhí)行。是中數(shù)據(jù)結(jié)束的標(biāo)示

正常情況下，一個(gè)流在執(zhí)行一次終端操作之后便結(jié)束了。本文通過(guò)復(fù)制流內(nèi)數(shù)據(jù)的方式，曲折的實(shí)現(xiàn)了同一個(gè)流上執(zhí)行多次操作。

Demo只是思路，其性能并不一定高效，尤其是數(shù)據(jù)都在內(nèi)存中處理時(shí)復(fù)制的開(kāi)銷(xiāo)很大。但如果流涉及大量I／O，也許性能會(huì)有提高。

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

此處重寫(xiě)了tryAdvance接口，只是簡(jiǎn)單的從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;
    }
}