实现未知大小的非并行 Spliterator？

Question

我对我的所有研究感到有点困惑。我有一个名为 TabularResultSet 的自定义界面（为了举例，我已经淡化了它）它遍历任何本质上是表格的数据集。它有一个类似于迭代器的 next() 方法，它可以循环遍历 QueryResultSet、来自剪贴板的选项卡式 table、CSV 等...

但是，我正在尝试创建一个环绕我的 TabularResultSet 并轻松将其转换为流的 Spliterator。我无法想象一种安全的并行化方法，因为 TabularResultSet 可能正在遍历 QueryResultSet，同时调用 next() 可能会造成严重破坏。我认为可以安全地完成并行化的唯一方法是让单个工作线程调用 next() 并将数据传递给并行线程以对其进行处理。

所以我认为并行化不是一个容易的选择。我怎样才能让这个东西在没有并行化的情况下流式传输？这是我到目前为止的工作...

public final class SpliteratorTest {

    public static void main(String[] args) {
       TabularResultSet rs = null; /* instantiate an implementation; */

       Stream<TabularResultSet> rsStream = StreamSupport.stream(new TabularSpliterator(rs), false);
    }

    public static interface TabularResultSet {
        public boolean next();

        public List<Object> getData();
    }

    private static final class TabularSpliterator implements Spliterator<TabularResultSet> {

        private final TabularResultSet rs;

        public TabularSpliterator(TabularResultSet rs) {
            this.rs = rs;
        }
        @Override
        public boolean tryAdvance(Consumer<? super TabularResultSet> action) {
            action.accept(rs);
            return rs.next();
        }

        @Override
        public Spliterator<TabularResultSet> trySplit() {
            return null;
        }

        @Override
        public long estimateSize() {
            return Long.MAX_VALUE;
        }

        @Override
        public int characteristics() {
            return 0;
        }
    }
}

Answer 1

你大部分时间都在那里。您现在要做的就是将您的 Spliterator 转换为 Stream。您可以使用 StreamSupport.stream(Spliterator, boolean) 方法来做到这一点。布尔参数是一个标志，表示您是否要进行并行流式处理（您可能想要 false，表示不并行）

如果您的 TabularResultSet 实现了 Iterator，您可以使用 Spliterators.spliteratorUnknownSize() 方法将 Iterator 转换为 Spliterator，它基本上执行您上面的代码所做的事情。

不确定是否值得添加特征，但您可能需要考虑 Spliterator.IMMUTABLE| Spliterator.ORDERED | Spliterator.NONNULL

祝你好运

Answer 2

扩展 Spliterators.AbstractSpliterator 可能是最简单的。如果你这样做，你只需要实现 tryAdvance。这可以变成并行流；并行性来自多次调用 tryAdvance 的流实现，对接收到的数据进行批处理，并在不同的线程中进行处理。

如果 TabularResultSet 类似于 JDBC ResultSet，我认为您不需要 Spliterator<TabularResultSet> 或 Stream<TabularResultSet>。相反，它看起来像 TabularResultSet 代表整个表格数据集，因此您可能希望每个拆分器或流元素代表 table 中的一行 - List<Object> 由 getData()?如果是这样，你会想要像下面这样的东西。

class TabularSpliterator extends Spliterators.AbstractSpliterator<List<Object>> {
    private final TabularResultSet rs;

    public TabularSpliterator(TabularResultSet rs) {
        super(...);
        this.rs = rs;
    }

    @Override public boolean tryAdvance(Consumer<? super List<Object>> action) {
        if (rs.next()) {
            action.accept(rs.getData());
            return true;
        } else {
            return false;
        }
    }
}

然后您可以通过调用 StreamSupport.stream().

将此拆分器的实例转换为流

注意：通常，Spliterator 实例不会从多个线程调用，甚至不需要是线程安全的。有关详细信息，请参阅“尽管...”开头的段落中的 Spliterator class documentation。

实现未知大小的非并行 Spliterator？

Implementing a non-parallel Spliterator for unknown size?

java

parallel-processing

lambda

java-8

spliterator