Java8流的复制

smartlion 发布于2019-08-15 13:25 / 1962人阅读

摘要：只是思路，其性能并不一定高效，尤其是数据都在内存中处理时复制的开销很大。方法将原始流中所有的数据添加到各个内，此处实现了复制此处重写了接口，只是简单的从中取出数据，执行。是中数据结束的标示

正常情况下，一个流在执行一次终端操作之后便结束了。本文通过复制流内数据的方式，曲折的实现了同一个流上执行多次操作。

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

文章版权归作者所有，未经允许请勿转载,若此文章存在违规行为，您可以联系管理员删除。

转载请注明本文地址：https://www.ucloud.cn/yun/68084.html

乐字节-Java8新特性之Stream流（上）

摘要：需要注意的是很多流操作本身就会返回一个流，所以多个操作可以直接连接起来，如下图这样，操作可以进行链式调用，并且并行流还可以实现数据流并行处理操作。为集合创建并行流。上一篇文章，小乐给大家介绍了《Java8新特性之方法引用》，下面接下来小乐将会给大家介绍Java8新特性之Stream，称之为流，本篇文章为上半部分。 1、什么是流? Java Se中对于流的操作有输入输出IO流,而Jav...

dingda 2019-08-19 11:35 评论0 收藏0
Java8特性②Stream简介

摘要：元素序列流也提供了一个接口，可以访问特定元素类型的一组有序值。因为集合是数据结构，所以它的主要目的是以特定的时间空间复杂度存储和访问元素如与。请注意，从有序集合生成流时会保留原有的顺序。由列表生成的流，其元素顺序与列表一致。流是什么流是Java API的新成员，它允许你以声明性方式处理数据集合（通过查询语句来表达，而不是临时编写一个实现）。可以把它们看成遍历数据集的高级迭代器。此外...

EasonTyler 2019-08-16 10:31 评论0 收藏0
《Java8实战》-第四章读书笔记（引入流Stream）

摘要：内部迭代与使用迭代器显式迭代的集合不同，流的迭代操作是在背后进行的。流只能遍历一次请注意，和迭代器类似，流只能遍历一次。流（Stream）流是什么流是Java API的新成员，它允许你以声明性方式处理数据集合（通过查询语句来表达，而不是临时编写一个实现）。就现在来说，你可以把它们看成遍历数据集的高级迭代器。此外，流还可以透明地并行处理，你无需写任何多线程代码了！我会在后面的笔记中...

_ivan 2019-08-19 11:01 评论0 收藏0
乐字节-Java8核心特性实战之Stream（流）

摘要：大家好，我是乐字节的小乐。需要注意的是很多流操作本身就会返回一个流，所以多个操作可以直接连接起来，如下图这样，操作可以进行链式调用，并且并行流还可以实现数据流并行处理操作。为集合创建并行流。大家好，我是乐字节的小乐。说起流，我们会联想到手机、电脑组装流水线,物流仓库商品包装流水线等等,如果把手机 ,电脑,包裹看做最终结果的话,那么加工商品前的各种零部件就可以看做数据源，而中间一系列的...

wenshi11019 2019-08-16 17:59 评论0 收藏0
用Java 8 的 Stream 来写代码，干净优雅！

摘要：补充一点使用数值流可以避免计算过程中拆箱装箱，提高性能。其目的主要是打开流，做出某种程度的数据映射过滤，然后返回一个新的流，交给下一个操作使用。终端操作的执行，才会真正开始流的遍历。 Java8的新特性主要是Lambda表达式和流，当流和Lambda表达式结合起来一起使用时，因为流申明式处理数据集合的特点，可以让代码变得简...

wemall 2021-11-22 15:24 评论0 收藏0