多线程和递归在一起

Question

我有以深度优先方式处理树结构的递归代码。代码基本上是这样的：

function(TreeNode curr) 
{
    if (curr.children != null && !curr.children.isEmpty()) 
    {
        for (TreeNode n : curr.children) 
    {
            //do some stuff
            function(n);
        }
    } 
    else 
    {
        //do some other processing
    }
}

我想使用线程来更快地完成这项工作。大部分时间都花在遍历上，所以我不想只创建一个线程来处理“其他处理”，因为它不需要那么长时间。我想我想在“做一些事情”时分叉线程，但这将如何工作？

Answer 1

Fork/Join framework 是一个很好的例子，它将被包含在 Java 7 中。作为用于 Java 6 的独立库，可以下载here。

类似这样的：

public class TreeTask extends RecursiveAction {
    private final TreeNode node;
    private final int level;

    public TreeTask(TreeNode node, int level) {
        this.node = node;
        this.level = leve;
    }

    public void compute() {
        // It makes sense to switch to single-threaded execution after some threshold
        if (level > THRESHOLD) function(node);

        if (node.children != null && !node.children.isEmpty()) {
            List<TreeTask> subtasks = new ArrayList<TreeTask>(node.children.size());
            for (TreeNode n : node.children) {
                // do some stuff
                subtasks.add(new TreeTask(n, level + 1));
            }
            invokeAll(subtasks); // Invoke and wait for completion
        } else {
            //do some other processing
        }
    }
}

...
ForkJoinPool p = new ForkJoinPool(N_THREADS);
p.invoke(root, 0);

fork/join 框架的关键点是工作窃取——在等待子任务完成的同时线程执行其他任务。它允许您以直接的方式编写算法，同时避免线程耗尽的问题，就像使用 ExecutorService 的幼稚方法一样。

Answer 2

在您在单个节点上工作的// do some stuff 代码块中，您可以做的是将节点提交给某种ExecutorService（以Runnable 的形式在节点上工作） .

您可以将ExecutorService 配置为由一定数量的线程池支持，允许您解耦“处理”逻辑（以及围绕创建线程、创建多少线程等的逻辑）来自您的树解析逻辑。

Answer 3

这个解决方案假设处理只发生在叶节点上，并且树的实际递归不需要很长时间。

我会让调用者线程进行递归，然后由 BlockingQueue 的工作人员通过线程池处理叶子。我这里有几个地方没有处理InterruptedException。

public void processTree(TreeNode top) {
    final LinkedBlockingQueue<Runnable> queue =
        new LinkedBlockingQueue<Runnable>(MAX_NUM_QUEUED);
    // create a pool that starts at 1 threads and grows to MAX_NUM_THREADS
    ExecutorService pool =
        new ThreadPoolExecutor(1, MAX_NUM_THREADS, 0L, TimeUnit.MILLISECONDS, queue,
            new RejectedExecutionHandler() {
                public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
                    queue.put(r);  // block if we run out of space in the pool
                }
            });
    walkTree(top, pool);
    pool.shutdown();
    // i think this will join with all of the threads
    pool.awaitTermination(WAIT_TILL_CHILDREN_FINISH_MILLIS, TimeUnit.MILLISECONDS);
}
private void walkTree(final TreeNode curr, ExecutorService pool) {
    if (curr.children == null || curr.children.isEmpty()) {
        pool.submit(new Runnable() {
            public void run() {
                processLeaf(curr);
            }
        });
        return;
    }
    for (TreeNode child : curr.children) {
        walkTree(child, pool);
    }
}
private void processLeaf(TreeNode leaf) {
    // ...
}