多线程和递归我一起

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

对于要包含在Java 7中的Fork/Join framework这是一个很好的例子。作为与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); 

叉的关键点/ join框架的工作窃取 - 在等待子任务线程完成执行其他任务。它可以让你以直接的方式编写算法，同时避免线程耗尽的问题，因为它可以与ExecutorService一样天真的应用程序。

Answer 2

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

您可以配置ExecutorService，您使用该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) { 
    // ... 
}