你的位置:首页 > Java教程

[Java教程]线程池学习


  新建一个线程并启动,开销会很大,因为运行线程需要的资源比调用对象方法需要的资源多得多。在很多情况下,线程被用于执行一类任务,而这类任务数量很多,发生的时间分布不均,如果为每个新任务都启用一个新线程来执行,则开销会太大,可以采用一种性能优化技术,就是使用线程池。

  将若干执行任务的线程放在池中,当有任务要执行时,从池中取出一个空闲线程来处理任务,处理完任务后,再讲线程对象放入池中。线程池实际上就是一个对象池,只是池中的对象都是线程。

  本文实例将实现一个线程池,可以给线程池分配任务,线程池中的线程自动获取任务并执行。

  关键技术:1.线程组ThreadGroup可以管理多个线程,所以让线程池继承ThreadGroup。

       2.无条件关闭线程池时,通过ThreadGroup的interrupt方法中断池中的所有线程。

       3.有条件关闭线程池时,通过ThreadGroup获得池中所有活动线程的引用,依次调用Thread的join方法等待活动线程执行完毕。当所有线程都运行结束时,线程池才                        能被关闭。

       4.将任务放在LinkedList中,由于LinkedList不支持同步,所以在添加任务和获取任务的方法声明中必须使用Synchronized关键字。

实例

package book.thread.pool;
/**
*定义任务的接口类
*/
public interface Task {
  public void perform() throws Exception;
}

package book.thread.pool;

public class MyTask implements Task{
  private int taskID = 0;//任务ID
  public MyTask(int id){
    this.taskID = id;
  }

  @Override
  public void perform() throws Exception {
    System.out.println("MyTask " + taskID + ":start");
    Thread.sleep(1000);
    System.out.println("MyTask " + taskID + ":end");
  }
}

package book.thread.pool;

import java.util.LinkedList;

public class MyThreadPool extends ThreadGroup{
  private boolean isAlive;//标志线程池是否开启
  private LinkedList taskQueue;//线程池中的任务队列
  private int threadID;//线程池中的线程ID
  private static int threadPoolID;//线程池ID
  //创建新的线程池,numThreads是池中的线程数
  public MyThreadPool(int numThreads){
    super("ThreadPool-"+(threadPoolID++));
    //设置该线程池的Daemon属性为true,表示当该线程池中的所有线程都被销毁时,该线程池会自动被销毁
    super.setDaemon(true);
    this.isAlive = true;
    this.taskQueue = new LinkedList();//新建一个任务队列
    //启动numThreads个工作线程
    for(int i = 0;i < numThreads; i++){
      new PooledThread().start();
    }
  }
  //添加新任务
  public synchronized void performTask(Task task){
    if(!this.isAlive){
      throw new IllegalStateException();//线程池被关闭,则抛出异常
    }
    if(task != null){
      this.taskQueue.add(task);//将任务放到任务队列的尾部
      notify();//通知工作线程取任务
    }
  }
  //获取任务
  protected synchronized Task getTask() throws InterruptedException{
    //如果任务列表为空,而且线程池没有被关闭,则继续等待任务
    while(this.taskQueue.size() == 0){
      if(!this.isAlive){
        return null;
      }
      wait();
    }
    //取任务列表的第一个任务
    return (Task)this.taskQueue.removeFirst();
  }
  //关闭线程池,所有线程停止,不再执行任务
  public synchronized void close(){
    if(isAlive){
      this.isAlive = false;
      this.taskQueue.clear();//清除任务
      this.interrupt();//中止线程池中的所有线程
    }
  }
  //关闭线程池,并等待线程池中的所有任务运行完成,但不能接收新任务
  public void join(){
    //通知其他等待线程“该线程池已关闭”的消息
    synchronized(this){
      isAlive = false;
      notifyAll();
    }
  //等待所有线程完成,首先建立一个新的线程组,activeCount方法获取线程池中活动线程的估计数
  Thread[] threads = new Thread[this.activeCount()];
  //将线程池中的活动线程拷贝到新创建的线程组threads中
  int count = this.enumerate(threads);
  for(int i = 0;i < count; i++){
    try {
      threads[i].join();//等待线程运行结束
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}
//内部类,用于执行任务的工作线程
private class PooledThread extends Thread{
  public PooledThread(){
    //第一个参数为该线程所在的线程组对象,即当前线程池对象
    //第二个参数为线程名字
    super(MyThreadPool.this,"PooledThread-" +(threadID++));
  }
  public void run(){
    //如果该线程没有被中止
    while(!isInterrupted()){
      //获取任务
      Task task = null;
    try {
      task = getTask();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    //只要线程池的任务列表不为空,getTask方法就总能得到一个任务
    //若getTask()返回null,则表示线程池中已经没有任务,而且线程池已经被关闭
    if(task == null){
      return;
    }
    //运行任务,捕捉异常
    try {
      task.perform();
     } catch (Exception e) {
      uncaughtException(this,e);
     }
    }
  }
  }
}

package book.thread.pool;

public class PoolTest {
  public static void main(String[] args) {
    int numThreads = 3;//线程池中的线程数
    MyThreadPool threadPool = new MyThreadPool(numThreads);//生成线程池
    int numTasks = 10;//任务数
    //运行任务
    for(int i = 0;i<numTasks;i++){
      threadPool.performTask(new MyTask(i));
    }
    //关闭线程池并等待所有任务完成
    threadPool.join();
  }
}

输出结果:

MyTask 0:start
MyTask 1:start
MyTask 2:start
MyTask 0:end
MyTask 3:start
MyTask 1:end
MyTask 4:start
MyTask 2:end
MyTask 5:start
MyTask 3:end
MyTask 6:start
MyTask 4:end
MyTask 7:start
MyTask 5:end
MyTask 8:start
MyTask 6:end
MyTask 9:start
MyTask 7:end
MyTask 8:end
MyTask 9:end

结果分析:MyThreadPool类是线程池的主体类,用于管理一组工作线程。

       1.继承ThreadGroup,可以使用ThreadGroup提供的方法管理线程池中的线程。

       2.performTask公有同步方法往线程池的任务队列中添加一个任务。如果线程池已被关闭,即isAlive属性为false,则不允许添加任务;添加任务后,调用notify方                         法,通知池中的工作线程取任务。

       3.getTask受保护同步方法从线程池的任务队列中获取一个任务。之所以声明为受保护的,是为了限制其他类的对象非法获取任务。如果任务队列中没有任务,则当                        前线程进入等待状态,如果线程池已被关闭,则直接返回null。

       4.close方法强制关闭线程池。通过ThreadGroup的interrupt方法中断线程池中所有运行的线程,清空任务队列,并且isAlive属性设置为false,表示不接收新任务

       5.join方法有条件的关闭线程池。isAlive属性置为false,表示线程池不再接收新任务,通过ThreadGroup获得正在运行的线程,通过Thread的join方法等待他们执                       行完任务后,再关闭线程池。

    PooledThread类是MyThreadPool的内部类,定义了工作线程,处于MyThreadPool线程池中。在run放在中不断的从线程池的任务队列中取任务,取到任务后,调用任务的perform方法执行任务。