並發遍歷二叉樹 Java 實現

單線程遍歷二叉樹是數據結構書中必講的算法, 但多線程訪問呢?

我最近寫了一個多線程遍歷二叉樹的算法, 當然既然是多線程,就不保證一定的順序, 只要節點能訪問到就算數.

算法的基本思路

1) 使用 java.util.concurrent.ThreadPoolExecutor 構造線程池

2) 當線程池有空余線程把樹的右子樹拆分給另外一個線程處理，　否則自己處理右子數

要考慮的情況

1) 如果二叉樹非常不平衡(左子樹很深,右子樹很淺), 會出現一個線程忙碌,而另外一個線程早結束的情況

工作線程過一定時間間隔就會檢查線程池是否有空閒線程, 有則分離當前節點的右子樹並委托給ThreadPool

2)判斷所有工作都結束

啟動工作任務時用 HashSet 記錄, 工作任務結束時清楚HashSet 記錄, 當HashSet size 為0 時,所有工作結束.

注意事項

1) 即使任務結束，ThreadPool 還是在運行的( waiting task ), 要調用 shutdown 通知 ThreadPool 結束

2) 主線程結束, 程序並不會結束, 想讓主線程等待 ThreadPool 結束, 要調用ThreadPool 的 awaitTermination 方法

程序

package com.pnp.javathreadpool;

import java.util.HashSet;
import java.util.Random;
import java.util.Stack;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ConcurTravBTree {
 class BTreeNode<T extends Comparable> {

  public BTreeNode<T> left;
  public BTreeNode<T> right;
  public T value;

  public BTreeNode(BTreeNode<T> left, BTreeNode<T> right, T value) {
   this.left = left;
   this.right = right;
   this.value = value;
  }

  public BTreeNode(T value) {
   this(null, null, value);
  }
 }

 public class BTree<T extends Comparable> {
  String dumpStr;

  public BTreeNode<T> root;

  public BTree() {
  }

  public BTree(BTreeNode<T> root) {
   this.root = root;
  }

  public void orderAdd(T v) {
   if (root == null)
    root = new BTreeNode(v);
   else
    orderAdd(root, v);
  }

  public void orderAdd(BTreeNode<T> node, T v) {
   if (node.value.compareTo(v) > 0) {
    if (node.left == null)
     node.left = new BTreeNode(v);
    else
     orderAdd(node.left, v);
   } else {
    if (node.right == null)
     node.right = new BTreeNode(v);
    else
     orderAdd(node.right, v);
   }
  }
 
  public void preOrderTraverse() {
   preOrderTraverse(root);
  }
 
  public void preOrderTraverse( BTreeNode<T> node) {
   if (node != null) {
    //System.out.println(node.value);
    try {
     Thread.sleep(100);
    } catch (InterruptedException e) {
     e.printStackTrace();
    }
    preOrderTraverse(node.left);
    preOrderTraverse(node.right);
   }
  } 
 }

 static int global_id = 0;
 public static final int POOLSIZE = 4;
 
 ThreadPoolExecutor threadPool;
 BTree<Integer> tree;
 HashSet<Integer> tasks = new HashSet<Integer>();
 

 public void init() {
  tree = new BTree<Integer>();

  /*
  int[] data = new int[] { 1, 3, 5 };
  for (int i = 0; i < data.length; i++)
   tree.orderAdd(data[i]);
  */
 
  Random r = new Random();
  for( int i=0; i< 500; i++) {
   tree.orderAdd( r.nextInt(500));
  }

  threadPool = new ThreadPoolExecutor(POOLSIZE, // corePoolSize
    POOLSIZE + 2, // maximumPoolSize
    60, // keepAliveTime
    TimeUnit.SECONDS, // keepAliveTime unit
    new ArrayBlockingQueue<Runnable>(POOLSIZE + 2), // workQueue
    new ThreadPoolExecutor.CallerRunsPolicy()); // handler
 }

 public void start() {
  threadPool.execute(new Visitor(tree.root));
 }

 class Visitor implements Runnable {
  private int id;
  private BTreeNode<Integer> node;

  public Visitor(BTreeNode<Integer> node) {
   this.id = ++global_id;
   this.node = node;
  }

  public boolean handleNode(BTreeNode<Integer> node) {
   if (node == null)
    return true;

   if (ConcurTravBTree.this.threadPool.getActiveCount() <= ConcurTravBTree.this.threadPool
     .getCorePoolSize()) {
    System.out.println("--- ActiveCount "
      + ConcurTravBTree.this.threadPool.getActiveCount());
    ConcurTravBTree.this.threadPool.execute(new Visitor(node));
    return true;
   }
   return false;
  }

  public void run() {
   System.out.println("[ job" + id + " ] ---start  ");
   tasks.add(id);
   if (handleNode(node.right)) // assign Node to a task
    node.right = null; //

   // pre-order traverse
   Stack<BTreeNode<Integer>> stack = new Stack<BTreeNode<Integer>>();
   while (node != null || !stack.isEmpty()) {
    long start = System.currentTimeMillis();
    while (node != null) {
     //System.out.println("[ job" + id + " ]" + node.value);
     try {
      Thread.sleep(100);
     } catch (InterruptedException e) {
      e.printStackTrace();
     }

     stack.push(node);
     node = node.left;
    }
    node = stack.pop();
    node = node.right;

    long cur = System.currentTimeMillis();
    if (cur - start > 1000) {
     if (handleNode(node)) {
      node = null;
     }
    }
   }
   System.out.println("[ job" + id + " ] ---end");
   tasks.remove(id);
   if ( tasks.size() == 0) {
    threadPool.shutdown();
   }
  }
 }

 public static void main(String[] args) {
  ConcurTravBTree tt = new ConcurTravBTree();
  System.out.println("init tree...");
  tt.init();

  System.out.println("start preOrderTraverse traverse...");
  long start = System.currentTimeMillis();
  tt.tree.preOrderTraverse();
  long end = System.currentTimeMillis();
  System.out.println("---- preOrderTraverse cost " + (end - start));
 
  System.out.println("start concurrent traverse...");
  long cstart = System.currentTimeMillis();
  tt.start();
  try {
   tt.threadPool.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
  } catch (InterruptedException e) {
   e.printStackTrace();
  }
  long cend = System.currentTimeMillis();
  System.out.println("---- current traverse cost " + (cend - cstart));
 }
}