我正在努力寻找一种适当的方法来从这个流中获得加速:
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;
}
}
但是我实在找不到好的解决办法。
最佳答案
要并行处理数据,您需要一个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 错误。
关于Java 并行流 : there's a way to navigate a binary tree?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/48148642/