线程池简单实现

转载地址：http://www.ibm.com/developerworks/cn/java/l-threadPool/#icomments

1. 任务接口（Task）:每个任务必须实现的接口，以供工作线程调度任务的执行。

/**
 * <p>Title: </p>
 * <p>Description: use to test thread pool</p>
 * <p>Copyright: Copyright (c) 2002</p>
 * <p>Company: </p>
 *
 * @author xingyong
 * 
 * @version 1.0
 */

public interface Task
{
   /**
    * set status of task .
    * @param flag
    */
   public void  setEnd(boolean flag) ;
   /**
    * start task
    * @throws java.lang.Exception
    */
   public abstract void startTask() throws Exception;

   /**
    * end tast
    * @throws java.lang.Exception
    */
   public abstract void endTask() throws Exception;

   /**
    * get status of task
    * @return boolean
    */
   public boolean isEnd() ;

}

2. 工作线程（WorkThread）: 线程池中线程

import java.io.PrintStream;
import java.util.Vector;
/**
 * <p>Title: </p>
 * <p>Description: use to test thread pool</p>
 * <p>Copyright: Copyright (c) 2002</p>
 * <p>Company: </p>
 *
 * @author xingyong
 * 
 * @version 1.0
 */

public class WorkThread extends Thread
{
   
   public int threadNum;
   private boolean flag;

   private Vector taskVector;
   private Task task;

   /**
    * @param vector
    * @param i
    */
   public WorkThread(java.util.Vector vector, int i)
   {
        flag = true;
        threadNum = i;
        taskVector = vector;
        //hide entry here
        super.start();
   }

   public void run()
   {
        while(flag && taskVector != null)
        {
            synchronized(taskVector)
            {
                while(taskVector.isEmpty() && flag)
                    try
                    {
                        taskVector.wait();
                    }
                    catch(Exception exception) { }
                try
                {
                    task = (Task)taskVector.remove(0);
                }
                catch(Exception ex)
                {
                    task = null;
                }
                if(task == null)
                    continue;
            }
            try
            {
                task.setEnd(false);
                task.startTask();
            }
            catch(Exception ex)
            {
                 ex.printStackTrace();
            }
            try
            {
                if(!task.isEnd())
                {
                    task.setEnd(true);
                    task.endTask();
                }
            }
            catch(Exception ex)
            {
                 ex.printStackTrace();
            }
          
        }//end of while
   }

   public void closeThread()
   {
        flag = false;
        try
        {
            if(task != null)
                task.endTask();
        }
        catch(Exception exception)
        {
            exception.printStackTrace();
        }
        synchronized(taskVector)
        {
            taskVector.notifyAll();
        }
   }
}

3. 线程池ThreadPool

/**
 * <p>Title: </p>
 * <p>Description: use to test thread pool</p>
 * <p>Copyright: Copyright (c) 2002</p>
 * <p>Company: </p>
 *
 * @author xingyong
 * * @version 1.0
 *note: this test program refers to  Tarak Modi's test structure
 */

public class TestThreadPool
{
  private ThreadPoolManager threadPoolManager = null;
  public  Object testlock = new Object();
  public  int count = 0 ;
  int jobsNumber = 200;
  int sizeOfThreadPool = 20;
  int debug = 2;
  public TestThreadPool(int threadPoolSize,int jobsNum)
  {
    //System.out.println("size:    "+sizeOfThreadPool+"    jobs:    "+jobsNumber);
    sizeOfThreadPool = threadPoolSize;
    jobsNumber = jobsNum;
    long timeNoPool = 0;
    long timePool = 0;
    count = 0;
    System.err.println("Begin of testing strategy -- no pool");
    long start = System.currentTimeMillis();
    try
    {
      for(int i=0; i<jobsNumber ; i++)
      {
      CalculationTaskTest calculationTask = new CalculationTaskTest();
      new Thread(calculationTask).start();
      }
    }
    catch (OutOfMemoryError er)
    {
      System.out.println("No pool :OutOfMemoryError");
    }
    catch (Exception ex)
    {
      ex.printStackTrace();
    }

    sleepToWait(5);
    if(debug > 3)
      System.out.println("no pool time:" +
		      (System.currentTimeMillis() - start));
      timeNoPool = System.currentTimeMillis() - start;
    System.err.println("End of no pool test");
    count = 0;
    //start = System.currentTimeMillis();
    System.err.println("Begin of  creating pool");
    threadPoolManager = new ThreadPoolManager(sizeOfThreadPool);
    System.err.println("End of  creating pool");
    System.err.println("Begin of testing the strategy  -- pool");
    //without the time for creating pool
    start = System.currentTimeMillis();
    try
    {
      for(int i=0; i<jobsNumber ; i++)
      {
      CalculationTaskTest calculationTaskTest = new CalculationTaskTest();
      threadPoolManager.addTask(calculationTaskTest);
      }
    }
    catch (OutOfMemoryError er)
    {
      System.out.println("pool :OutOfMemoryError"+" "+sizeOfThreadPool+" "+jobsNumber);
    }
    catch (Exception ex)
    {
      ex.printStackTrace();
    }

    sleepToWait(5);
    if(debug > 3)
      System.out.println("pool time:" +
		       (System.currentTimeMillis() - start));
    timePool = System.currentTimeMillis() - start;
    System.err.println("End of thread pool test");
    System.out.println("without pool: "+timeNoPool+
    "    with pool: "+timePool);


    System.exit(0);

  }
  public  void sleepToWait(long l)
  {
   while(true)
   {
     synchronized(testlock)
       {
          if( count == jobsNumber )
             return;
        }
     try
      {
        //you can change it to wait end of all tasks
        Thread.sleep(l);
      }
      catch (Exception ex)
      { }
    }
  }

  public static void main(String[] args)
  { int poolSize = 20;
    int jobs= 200;

    if( args.length < 2 )
    {
			  System.err.println("Usage: java TestThreadPool " +
			    "<Size of ThreadPool> <jobs> ");
			  System.exit(-1);
    }
    try
    {
      poolSize = Integer.parseInt(args[0]);
      jobs = Integer.parseInt(args[1]);
    }
    catch (Exception ex)
    {
      System.out.println("Please input integer.");
      System.exit(-1);
    }
    new TestThreadPool(poolSize,jobs);
  }
  private  class CalculationTaskTest
                 implements Task,Runnable
{
  private boolean isEnd = true;
  public CalculationTaskTest()
  {
   isEnd = true;
  }

  public void setEnd(boolean flag)
  {
    /**@todo: Implement this Task method*/
    //throw new java.lang.UnsupportedOperationException("Method setEnd() not yet implemented.");
    isEnd = flag;
  }
  public void run()
  {
    int i = 1;
    int r = 1;



    //System.err.println("r:"+r);
     try
      {
        //you can change this line  to simulate something
        for(int ii=0;ii<100;ii++)
           r = r+i*i;
      }
      catch (Exception ex)
      { }
    synchronized(testlock)
    {
				 count++;

     }

  }
  public void startTask() throws Exception
  {
    /**@todo: Implement this Task method*/
    //throw new java.lang.UnsupportedOperationException("Method startTask() not yet implemented.");
    //System.err.print("use pool");
    run();
  }
  public void endTask() throws Exception
  {
    /**@todo: Implement this Task method*/
    //throw new java.lang.UnsupportedOperationException("Method endTask() not yet implemented.");
    //free resource here
  }
  public boolean isEnd()
  {
    /**@todo: Implement this Task method*/
    //throw new java.lang.UnsupportedOperationException("Method isEnd() not yet implemented.");
    return isEnd;
  }
}
}

4. 线程池管理器（ThreadPoolManager）:用于创建并管理线程池

import java.util.Vector;
/**
 * <p>Title: </p>
 * <p>Description: use to test simple thread pool</p>
 * <p>Copyright: Copyright (c) 2002</p>
 * <p>Company: </p>
 *
 * @author xingyong
 *
 * @version 1.0
 */
public class ThreadPoolManager
{

   /**
    * the number of threads in pool
    */
   private int threadNum;
   /**
    * the default number of threads in pool
    */
   private int defaultThreadNum;
   /**
    * the vector of threads in pool
    */
   private Vector workThreadVector;
   /**
    * the vector of tasks
    */
   private Vector taskVector;

   /**
    * @param i
    */
   public ThreadPoolManager(int i)
   {
        taskVector = new Vector(10, 10);
        //initial thread number
        defaultThreadNum = 10;
        if(i > 0)
            defaultThreadNum = i;
        //call thread
        CreateThreadPool(i);
   }

   public ThreadPoolManager()
   {
        this(10);
   }
   /**
    *
    * @return
    */
   public boolean isAllTaskFinish()
   {
     return taskVector.isEmpty();
   }
   /**
    * @return int
    */
   public int getThreadNum()
   {
        return threadNum;
   }



   /**
    * create thread pool
    * @param i
    */
   private void CreateThreadPool(int i)
   {
        if(workThreadVector == null)
            workThreadVector = new Vector(i);
        Object obj = null;
        //create threads
        synchronized(workThreadVector)
        {
            for(int j = 0; j < i; j++)
            {
                threadNum++;
                WorkThread workThread = new WorkThread(taskVector, threadNum);
                workThreadVector.addElement(workThread);
            }

        }
   }

   /**
    * add task to task vector and notify work Threads in pool
    * to do it
    * @param taskObj
    */
   public void addTask(Task taskObj)
   {
        if(taskObj == null)
            return;
        synchronized(taskVector)
        {
            taskVector.addElement(taskObj);
            taskVector.notifyAll();
        }
   }

  /**
   * destroy  threads in pool
   */
   public void closeThread()
   {
        Object obj = null;

        while(!workThreadVector.isEmpty())
        {


            try
            {
                WorkThread workThread = (WorkThread)workThreadVector.remove(0);
                workThread.closeThread();
                continue;
            }
            catch(Exception exception)
            {

                    exception.printStackTrace();
            }
            break;
        }
   }
}