Java 并行流 : there's a way to navigate a binary tree?

Question

我正在努力寻找一种适当的方法来从这个流中获得加速:

    StreamSupport.stream(new BinaryTreeSpliterator(root), true)
                .parallel()
                .map(node -> processor.onerousFunction(node.getValue()))
                .mapToInt(i -> i.intValue())
                .sum()

onerousFunction() 只是一个使线程工作一段时间并返回节点的 int 值的函数。

无论我使用多少 cpu，执行时间始终保持不变。我认为问题出在我写的 Spliterator 中:

    public class BinaryTreeSpliterator extends AbstractSpliterator<Node> {

        private LinkedBlockingQueue<Node> nodes = new LinkedBlockingQueue<>();

        public BinaryTreeSpliterator(Node root) {
            super(Long.MAX_VALUE, NONNULL | IMMUTABLE);
            this.nodes.add(root);
        }

        @Override
         public boolean tryAdvance(Consumer<? super Node> action) {
            Node current = this.nodes.poll();
            if(current != null) {
                action.accept(current);
                if(current.getLeft() != null) 
                    this.nodes.offer(current.getLeft());
                if(current.getRight() != null)
                    this.nodes.offer(current.getRight());
                return true;
            }
            return false;
        }

    }

但是我实在找不到好的解决办法。

Answer 1

要并行处理数据，您需要一个trySplit 实现来将部分数据作为新的Spliterator 实例返回。每个拆分器实例都由一个线程遍历。因此，顺便说一下，您不需要在拆分器中使用线程安全的集合。但是您的问题是您从 AbstractSpliterator 继承了 trySplit 实现，尽管对您的数据一无所知，它确实尝试提供一些并行支持。

它是这样做的，通过顺序请求一些项目，将它们缓冲到一个数组中并返回一个新的基于数组的拆分器。不幸的是，它不能很好地处理“未知大小”(这同样适用于一般的并行流实现)。默认情况下它将缓冲 1024 个元素，如果有那么多元素，下次缓冲更多。更糟糕的是，流实现不会使用基于数组的拆分器的良好拆分功能，因为它将“未知大小”视为文字 Long.MAX_VALUE，得出的结论是您的拆分器具有比 1024 个元素多得多的元素因此，在数组中，甚至不会尝试拆分基于数组的拆分器。

您的拆分器可以实现更合适的 trySplit 方法:

public class BinaryTreeSpliterator extends AbstractSpliterator<Node> {
    /**
     * a node that has not been traversed, but its children are only
     * traversed if contained in this.pending
     * (otherwise a different spliterator might be responsible)
     */
    private Node pendingNode;
    /** pending nodes needing full traversal */
    private ArrayDeque<Node> pending = new ArrayDeque<>();

    public BinaryTreeSpliterator(Node root) {
        super(Long.MAX_VALUE, NONNULL | IMMUTABLE);
        push(root);
    }

    private BinaryTreeSpliterator(Node pending, Node next) {
        super(Long.MAX_VALUE, NONNULL | IMMUTABLE);
        pendingNode = pending;
        if(next!=null) this.pending.offer(next);
    }
    private void push(Node n) {
        if(pendingNode == null) {
            pendingNode = n;
            if(n != null) {
                if(n.getRight()!=null) pending.offerFirst(n.getRight());
                if(n.getLeft() !=null) pending.offerFirst(n.getLeft());
            }
        }
        else pending.offerFirst(n);
    }

    @Override
     public boolean tryAdvance(Consumer<? super Node> action) {
        Node current = pendingNode;
        if(current == null) {
            current = pending.poll();
            if(current == null) return false;
            push(current.getRight());
            push(current.getLeft());
        }
        else pendingNode = null;
        action.accept(current);
        return true;
    }

    @Override
    public void forEachRemaining(Consumer<? super Node> action) {
        Node current = pendingNode;
        if(current != null) {
            pendingNode = null;
            action.accept(current);
        }
        for(;;) {
            current = pending.poll();
            if(current == null) break;
            traverseLocal(action, current);
        }
    }
    private void traverseLocal(Consumer<? super Node> action, Node current) {
        do {
            action.accept(current);
            Node child = current.getLeft();
            if(child!=null) traverseLocal(action, child);
            current = current.getRight();
        } while(current != null);
    }

    @Override
    public Spliterator<Node> trySplit() {
        Node next = pending.poll();
        if(next == null) return null;
        if(pending.isEmpty()) {
            pending.offer(next);
            next = null;
        }
        if(pendingNode==null) return next==null? null: new BinaryTreeSpliterator(next);
        Spliterator<Node> s = new BinaryTreeSpliterator(pendingNode, next);
        pendingNode = null;
        return s;
    }
}

请注意，此拆分器也符合 ORDERED 拆分器的条件，保持从左上到右的顺序。完全无序的拆分器可以稍微简单地实现。

您可以实现比继承的默认方法更有效的 forEachRemaining 方法，例如

@Override
public void forEachRemaining(Consumer<? super Node> action) {
    Node current = pendingNode;
    if(current != null) {
        pendingNode = null;
        action.accept(current);
    }
    for(;;) {
        current = pending.poll();
        if(current == null) break;
        traverseLocal(action, current);
    }
}
private void traverseLocal(Consumer<? super Node> action, Node current) {
    do {
        action.accept(current);
        Node child = current.getLeft();
        if(child!=null) traverseLocal(action, child);
        current = current.getRight();
    } while(current != null);
}

但是如果您的应用程序必须处理不平衡的树(具体而言，本例中的左路径很长)，则此方法可能会导致 stackoverflow 错误。