Vincent

設計目標
?????提供一個線程池的組件，具有良好的伸縮性，當線程夠用時，銷毀不用線程，當線程不夠用時，自動增加線程數量；
?????提供一個工作任務接口和工作隊列，實際所需要的任務都必須實現這個工作任務接口，然后放入工作隊列中；
?????線程池中的線程從工作隊列中，自動取得工作任務，執行任務。
主要控制類和功能接口設計
線程池管理器?ThreadPoolManager?的功能：
?????管理線程池中的各個屬性變量
ü????最大工作線程數
ü????最小工作線程數
ü????激活的工作線程總數
ü????睡眠的工作線程總數
ü????工作線程總數?(即：激活的工作線程總數+睡眠的工作線程總數)
?????創建工作線程
?????銷毀工作線程
?????啟動處于睡眠的工作線程
?????睡眠處于激活的工作線程
?????縮任務：當工作線程總數小于或等于最小工作線程數時，銷毀多余的睡眠的工作線程，使得現有工作線程總數等于最小工作任務總數
?????伸任務：當任務隊列任務總數大于工作線程數時，增加工作線程總數至最大工作線程數
?????提供線程池啟動接口
?????提供線程池銷毀接口
工作線程?WorkThread??的功能：
?????從工作隊列取得工作任務
?????執行工作任務接口中的指定任務
工作任務接口?ITask???的功能：
?????提供指定任務動作
工作隊列?IWorkQueue??的功能：
?????提供獲取任務接口，并刪除工作隊列中的任務；
?????提供加入任務接口；
?????提供刪除任務接口；
?????提供取得任務總數接口；
?????提供自動填任務接口；（當任務總數少于或等于默認總數的25％時，自動裝填）
?????提供刪除所有任務接口；


Code


ThreadPoolManager：
=====================================

CODE:

---

package test.thread.pool1;
import java.util.ArrayList;
import java.util.List;
import test.thread.pool1.impl.MyWorkQueue;

/**
 * <p>Title: 線程池管理器</p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2005</p>
 * <p>Company: </p>
 * @author not attributable
 * @version 1.0
 */

public class ThreadPoolManager {
  /*最大線程數*/
  private int threads_max_num;

  /*最小線程數*/
  private int threads_min_num;
  
  /* 線程池線程增長步長 */
  private int threads_increase_step = 5;

  /* 任務工作隊列 */
  private IWorkQueue queue;
  
  /* 線程池監視狗 */
  private PoolWatchDog poolWatchDog ;
  
  /* 隊列線程 */
  private Thread queueThread ;
  
  /* 線程池 封裝所有工作線程的數據結構 */
  private List pool = new ArrayList();
  
  /* 線程池中 封裝所有鈍化后的數據結構*/
  private List passivePool = new ArrayList();
  
  /* 空閑60秒 */
  private static final long IDLE_TIMEOUT = 60000L;
  
  /* 關閉連接池標志位 */
  private boolean close = false;
  
  /**
   * 線程池管理器
   * @param queue 任務隊列
   * @param threads_min_num 工作線程最小數
   * @param threads_max_num 工作線程最大數
   */
  public ThreadPoolManager(int threads_max_num
                          ,int threads_min_num
                          ,IWorkQueue queue){
    this.threads_max_num = threads_max_num;
    this.threads_min_num = threads_min_num;
    this.queue = queue;    
  }

  /**
   * 線程池啟動
   */
  public void startPool(){
    System.out.println("=== startPool..........");
    poolWatchDog = new PoolWatchDog("PoolWatchDog");
    poolWatchDog.setDaemon(true);
    poolWatchDog.start();
    System.out.println("=== startPool..........over");
  }

  /**
   * 線程池銷毀接口
   */
  public void destoryPool(){
    System.out.println("==========================DestoryPool starting ...");
    this.close = true;
    int pool_size = this.pool.size();
    
    //中斷隊列線程
    System.out.println("===Interrupt queue thread ... ");
    queueThread.interrupt();
    queueThread = null;
    
    System.out.println("===Interrupt thread pool ... ");
    Thread pool_thread = null;
    for(int i=0; i<pool_size; i++){
      pool_thread = (Thread)pool.get(i);
      if(pool_thread !=null 
      && pool_thread.isAlive() 
      && !pool_thread.isInterrupted()){
        pool_thread.interrupt();
        System.out.println("Stop pool_thread:"
                          +pool_thread.getName()+"[interrupt] "
                          +pool_thread.isInterrupted());
      }
    }//end for
    
    if(pool != null){
      pool.clear();
    }
    if(passivePool != null){
      pool.clear();
    }
    
    try{
      System.out.println("=== poolWatchDog.join() starting ...");
      poolWatchDog.join();
      System.out.println("=== poolWatchDog.join() is over ...");
    }
    catch(Throwable ex){
      System.out.println("###poolWatchDog ... join method throw a exception ... "
                          +ex.toString());
    }
    
    poolWatchDog =null;
    System.out.println("==============================DestoryPool is over ...");    
  }
  
  
  public static void main(String[] args) throws Exception{
    ThreadPoolManager threadPoolManager1 = new ThreadPoolManager(10,5,new MyWorkQueue(50,30000));
    
    threadPoolManager1.startPool();
    Thread.sleep(60000);
    threadPoolManager1.destoryPool();
  }
  
  /**
   * 線程池監視狗
   */
  private class PoolWatchDog extends Thread{
    public PoolWatchDog(String name){
      super(name);
    }
  
    public void run(){
      Thread workThread = null;
      Runnable run = null;
      
      //開啟任務隊列線程，獲取數據--------
      System.out.println("===QueueThread starting ... ... ");
      queueThread = new Thread(new QueueThread(),"QueueThread");
      queueThread.start();
      
      System.out.println("===Initial thread Pool ... ...");
      //初始化線程池的最小線程數,并放入池中
      for(int i=0; i<threads_min_num; i++){
        run = new WorkThread();
        workThread = new Thread(run,"WorkThread_"+System.currentTimeMillis()+i);
        workThread.start();
        if(i == threads_min_num -1){
          workThread = null;
          run = null;
        }
      }
      System.out.println("===Initial thread Pool..........over ,and get pool's size:"+pool.size());

      //線程池線程動態增加線程算法--------------
      while(!close){
      
        //等待5秒鐘，等上述線程都啟動----------
        synchronized(this){          
          try{
            System.out.println("===Wait the [last time] threads starting ....");
            this.wait(15000);
          }
          catch(Throwable ex){
            System.out.println("###PoolWatchDog invoking is failure ... "+ex);
          }
        }//end synchronized
          
        //開始增加線程-----------------------spread動作
        int queue_size = queue.getTaskSize();
        int temp_size = (queue_size - threads_min_num);
        
        if((temp_size > 0) && (temp_size/threads_increase_step > 2) ){
          System.out.println("================Spread thread pool starting ....");
          for(int i=0; i<threads_increase_step && (pool.size() < threads_max_num); i++){
            System.out.println("=== Spread thread num : "+i);
            run = new WorkThread();
            workThread = new Thread(run,"WorkThread_"+System.currentTimeMillis()+i);
            workThread.start();
          }//end for
          
          workThread = null;
          run = null;    
          System.out.println("===Spread thread pool is over .... and pool size:"+pool.size());
        }//end if
          
        //刪除已經多余的睡眠線程-------------shrink動作
        int more_sleep_size = pool.size() - threads_min_num;//最多能刪除的線程數
        int sleep_threads_size = passivePool.size();
        if(more_sleep_size >0 && sleep_threads_size >0){
          System.out.println("================Shrink thread pool starting ....");        
          for(int i=0; i < more_sleep_size && i < sleep_threads_size ; i++){
            System.out.println("=== Shrink thread num : "+i);
            Thread removeThread = (Thread)passivePool.get(0);
            if(removeThread != null && removeThread.isAlive() && !removeThread.isInterrupted()){
              removeThread.interrupt();
            }
          }
          System.out.println("===Shrink thread pool is over .... and pool size:"+pool.size());          
        }

        System.out.println("===End one return [shrink - spread operator] ....");    
      }//end while
    }//end run 
  }//end private class
  
  /**
   * 工作線程
   */
  class WorkThread implements Runnable{
  
    public WorkThread(){
    }
  
    public void run(){
      String name = Thread.currentThread().getName();
      System.out.println("===Thread.currentThread():"+name);
      pool.add(Thread.currentThread());    
    
      while(true){
      
        //獲取任務---------
        ITask task = null;
        try{
          System.out.println("===Get task from queue is starting ... ");
          //看線程是否被中斷，如果被中斷停止執行任務----
          if(Thread.currentThread().isInterrupted()){
            System.out.println("===Breaking current thread and jump whlie [1] ... ");
            break;
          }
          task = queue.getTask();
        }
        catch(Throwable ex){
          System.out.println("###No task in queue:"+ex);
        }//end tryc
        
        if(task != null){
          //執行任務---------
          try{
            System.out.println("===Execute the task is starting ... ");
            //看線程是否被中斷，如果被中斷停止執行任務----
            if(Thread.currentThread().isInterrupted()){
              System.out.println("===Breaking current thread and jump whlie [1] ... ");
              break;
            }     
            task.executeTask();
            //任務執行完畢-------
            System.out.println("===Execute the task is over ... ");
          }
          catch(Throwable ex){
            System.out.println("###Execute the task is failure ... "+ex);
          }//end tryc
          
        }else{
          //沒有任務，則鈍化線程至規定時間--------
          synchronized(this){
            try{
              System.out.println("===Passivate into passivePool ... ");
              
              //看線程是否被中斷，如果被中斷停止執行任務----
              boolean isInterrupted = Thread.currentThread().isInterrupted();
              if(isInterrupted){
                System.out.println("===Breaking current thread and jump whlie [1] ... ");
                break;
              }
//              passivePool.add(this);
            passivePool.add(Thread.currentThread());

              
              //準備睡眠線程-------
              isInterrupted = Thread.currentThread().isInterrupted();
              if(isInterrupted){
                System.out.println("===Breaking current thread and jump whlie [2] ... ");
                break;
              }              
              this.wait(IDLE_TIMEOUT);
            }
            catch(Throwable ex1){
              System.out.println("###Current Thread passivate is failure ... break while cycle. "+ex1);
              break;
            }
          }          
        }        
      }//end while--------
      
      if(pool.contains(passivePool)){
        pool.remove(this);
      }
      if(passivePool.contains(passivePool)){
        passivePool.remove(this);
      }
      System.out.println("===The thread execute over ... "); 
    }//end run----------
  }
  
  
  class QueueThread implements Runnable{
  
    public QueueThread(){
    }
  
    public void run(){
      while(true){
        //自動裝在任務--------
        queue.autoAddTask();
        System.out.println("===The size of queue's task is "+queue.getTaskSize());
      
        synchronized(this){
          if(Thread.currentThread().isInterrupted()){
            break;
          }else{
              try{
                this.wait(queue.getLoadDataPollingTime());
              }
              catch(Throwable ex){
                System.out.println("===QueueThread invoked wait is failure ... break while cycle."+ex);
                break;
              }
          }//end if
        }//end synchr
        
      }//end while
    }//end run
  } 
}

---

WorkQueue
=====================================

CODE:

---

package test.thread.pool1;

import java.util.LinkedList;
import test.thread.pool1.impl.MyTask;

/**
 * <p>Title: 工作隊列對象 </p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2005</p>
 * <p>Company: </p>
 * @author not attributable
 * @version 1.0
 */

public abstract class WorkQueue implements IWorkQueue{
  /* 預計裝載量 */
  private int load_size;
  
  /* 數據裝載輪循時間 */
  private long load_polling_time;
  
  /* 隊列 */
  private LinkedList queue = new LinkedList();
  
  /**
   * 
   * @param load_size 預計裝載量
   * @param load_polling_time 數據裝載輪循時間
   */
  public WorkQueue(int load_size,long load_polling_time){
    this.load_size = (load_size <= 10) ? 10 : load_size;
    this.load_polling_time = load_polling_time;
  }

  /* 數據裝載輪循時間 */
  public long getLoadDataPollingTime(){
    return this.load_polling_time;
  }


  /*獲取任務，并刪除隊列中的任務*/
  public synchronized ITask getTask(){
    ITask task = (ITask)queue.getFirst();
    queue.removeFirst();
    return task;
  }

  /*加入任務*/
  public void  addTask(ITask task){
    queue.addLast(task);
  }

  /*刪除任務*/
  public synchronized void removeTask(ITask task){
    queue.remove(task);
  }

  /*任務總數*/
  public synchronized int getTaskSize(){
    return queue.size();
  }

  /*自動裝填任務*/
  public synchronized void autoAddTask(){
  
    synchronized(this){
      float load_size_auto = load_size - getTaskSize() / load_size;
      System.out.println("===load_size_auto:"+load_size_auto);
      
      if(load_size_auto > 0.25){        
        autoAddTask0();
      }
      else {
        System.out.println("=== Not must load new work queue ... Now! ");
      }    
    }
  }

  /*刪除所有任務*/
  public synchronized void clearAllTask(){
    queue.clear();
  }
  
  /**
   * 程序員自己實現該方法
   */
  protected abstract void autoAddTask0();
}

---

MyWorkQueue
=====================================

CODE:

---

package test.thread.pool1.impl;

import java.util.LinkedList;
import test.thread.pool1.WorkQueue;

/**
 * <p>Title: 例子工作隊列對象 </p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2005</p>
 * <p>Company: </p>
 * @author not attributable
 * @version 1.0
 */

public class MyWorkQueue extends WorkQueue{

  /**
   * @param load_size 預計裝載量
   * @param load_polling_time 數據裝載輪循時間
   */
  public MyWorkQueue(int load_size,long load_polling_time){
    super(load_size,load_polling_time);
  }

  /**
   * 自動加載任務
   */
  protected synchronized void autoAddTask0(){
    //-------------------
    System.out.println("===MyWorkQueue ...  invoked autoAddTask0() method ...");
    for(int i=0; i<10; i++){
      System.out.println("===add task :"+i);
      this.addTask(new MyTask());
    }    
    //-------------------
  }
}

---

MyTask
=====================================

CODE:

---

package test.thread.pool1.impl;
import test.thread.pool1.ITask;

/**
 * <p>Title: 工作任務接口 </p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2005</p>
 * <p>Company: </p>
 * @author not attributable
 * @version 1.0
 */

public class MyTask implements ITask {

  /**
   * 執行的任務
   * @throws java.lang.Throwable
   */
  public void executeTask() throws Throwable{
    System.out.println("["+this.hashCode()+"] MyTask ... invoked executeTask() method ... ");
  }
}