工作生活

Java 8 之Stream Spliterator

2018-08-02  本文已影响0人  Unyielding_L

定义

数据源

处理数据源的方式

特征 characteristics

Spliterator 还声明了 一组关于它的结构和源的特征(characteristics)，包含以下以下几种：

Tips

并行的实现

尽管在并行算法中有明显的实用功能，但spliterator并不向我们期望的那样是线程安全的;相反，使用spliterator的并行算法的实现应该确保spliterator一次只使用一个线程。这个通常很容易通过 串行线程封闭 来实现:通常使用递归分解这个经典的并行算法。调用{@link #trySplit()}的线程可以将返回的Spliterator传递给另一个线程，而这个线程又可以遍历或进一步拆分这个Spliterator。如果两个或多个线程在同一个Spliterator上同时操作，则不定义分割和遍历的行为。如果原始线程将一个spliterator传递给另一个线程进行处理，那么最好是在使用{@link #tryAdvance(Consumer) tryAdvance()}的任何元素之前进行切换，因为某些保证(例如{@link #estimateSize()}对于{@code size}spliterator的精度)只有在遍历开始之前才有效。


Spliterator分割图.png

Spliterator通过支持分割和单元素迭代，除了支持串行遍历，还支持高效的并行遍历。另外，Spliterator 不像Iterator设计的那样设计两个方法hasNext 判断是否有元素和next() 返回元素进行消费，Spliterator 设计一个tryAdvance方法，消费元素，如果有就消费并返回true，如果没有则返回false，不需要两个独立的方法。
对于可变源，如果在Spliterator绑定到其数据源和遍历结束之间对源进行结构上的干扰(添加、替换或删除元素)，可能会出现随机和不确定的影响。
对于structurally interfered 可以有一下几个方法避免：

/**
 * @Author unyielding
 * @date 2018/7/26 0026 19:48
 * @desc 一个类(除了当做例子之外，它不是一个非常有用的类)，
 * 它维护一个数组， 其中实际数据保存在偶数位置，而不相关的标记数据保存在奇数位置。
 * 它的Spliterator会忽略标记数据。
 */
public class TaggedArray<T> {
    private final Object[] elements;//创建后，不可变的
    /**
     * 构造方法
     *
     * @param data 实际数据
     * @param tags 标记数据
     */
    TaggedArray(T[] data, Object[] tags) {
        int size = data.length;
        //保证实际数据数组和标记数据数组的大小相同
        if (tags.length != size) throw new IllegalArgumentException();
        this.elements = new Object[2 * size];
        //初始化elements 数组
        for (int i = 0, j = 0; i < size; ++i) {
            elements[j++] = data[i];
            elements[j++] = tags[i];
        }
    }

    public Spliterator<T> spliterator() {
        return new TaggedArraySpliterator<>(elements, 0, elements.length);
    }

    static class TaggedArraySpliterator<T> implements Spliterator<T> {
        private final Object[] array;

        private int origin; //当前索引，在分割或者遍历时使用

        private final int fence;//最大的下标加一

        TaggedArraySpliterator(Object[] array, int origin, int fence) {
            this.array = array;
            this.origin = origin;
            this.fence = fence;
        }

        /**
         *  批量遍历
         * @param action 消费函数 {@link Consumer} 的子类，可以通过lambda表达式表示
         */
        @Override
        public void forEachRemaining(Consumer<? super T> action) {
            for (; origin < fence; origin += 2) {
                action.accept((T) array[origin]);
            }
        }

        /**
         *  处理单个元素
         * @param action 消费函数 {@link Consumer} 的子类，可以通过lambda表达式表示
         * @return 如果有元素消费就返回true,如果没有就直接返回false
         */
        @Override
        public boolean tryAdvance(Consumer<? super T> action) {
            if (origin < fence) {
                action.accept((T) array[origin]);
                origin += 2;
                return true;
            }
            return false;
        }

        /**
         * 分割数据源
         * @return 返回分割后生成的Spliterator
         */
        @Override
        public Spliterator<T> trySplit() {
            int lo = origin;
            int mid = ((lo + fence) >> 1) & 1;//强制中点数为偶数
            if (lo < mid) {
                origin = mid;//重置Spliterator的 当前下标
                return new TaggedArraySpliterator<>(array, lo, mid);
            }//太小不需要拆分
            return null;
        }

        /**
         * 估计剩余还有多少元素
         * @return 剩余还有多少元素
         */
        @Override
        public long estimateSize() {
            return (long) ((fence - origin) / 2);
        }

        /**
         * 获取特征值 用户可以根据 特征值 ，
         * 用户可以根据 配置更好的控制和优化它的使用
         * @return
         */
        @Override
        public int characteristics() {
            return ORDERED | IMMUTABLE | SIZED | SUBSIZED;
        }
    }

    /**
     * 并行遍历
     * @param a 一个{@link TaggedArray} 实例
     * @param action
     * @param <T> 每个元素的值
     */
    static <T> void parEach(TaggedArray<T> a, Consumer<T> action) {
        Spliterator<T> spliterator = a.spliterator();
        long targetBatchSize = spliterator.estimateSize()
                / (ForkJoinPool.getCommonPoolParallelism() * 8);
        new ParEach<>(null, spliterator, action, targetBatchSize).invoke();
    }
}

并行计算器 ，其实就是继承CountedCompleter 一个可以放到forlk/join 线程池里的类

    /**
     * 并行计算器
     * @param <T> 元素的类型
     */
    static class ParEach<T> extends CountedCompleter<T> {
        final Spliterator<T> spliterator;
        final Consumer<T> action;
        final long targetBatchSize;

        ParEach(ParEach<T> parent, Spliterator<T> spliterator,
                Consumer<T> action, long targetBatchSize) {
            super(parent);
            this.spliterator = spliterator;
            this.action = action;
            this.targetBatchSize = targetBatchSize;
        }

        @Override
        public void compute() {
            Spliterator<T> sub;
            while (spliterator.estimateSize() > targetBatchSize
                    && (sub = spliterator.trySplit()) != null) {
                addToPendingCount(1);
                new ParEach<>(this, sub, action, targetBatchSize).fork();
            }
            spliterator.forEachRemaining(action);
            propagateCompletion();
        }
    }

至于生成Stream 的姿势 详见Java 之Stream 生成姿势
代码地址
推荐阅读
https://www.jianshu.com/p/af22a9d8ce98

上一篇下一篇

猜你喜欢

热点阅读