多線程和遞歸我一起

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) { 
    // ... 
}