JAVA模拟线程池

       线程池就像数据库连接池一样，是一个对象池。所有的线程对象都有一个共同的目的，那就是为了提高对象的使用率，从而达到提高程序效率的目的。比如对于Servlet，它被设计为多线程的(如果它是单线程的，你就可以想象，当1000个人同时请求一个页面时，在第一个人获得请求结果之前，其它999个人都在郁闷地等待)，如果为每个用户的每一次请求都创建一个新的线程对象来运行的话，系统就会在创建线程和销毁线程上耗费很大的开销，大大降低系统的效率。因此，Servlet多线程机制背后有一个线程池在支持，线程池在初始化初期就创建了一定数量的线程对象，通过提高对这些对象的利用率，避免高频率地创建对象，从而达到提高程序的效率的目的。

         在本文中将实现一个最简单的线程池，从中理解它的实现原理。为此我们定义了四个类，它们的用途及具体实现如下：

Task(任务)

 

/**
 *  这是代表任务的抽象类，定义了一个deal方法，继承Task抽象类的子类
 *  需要实现这个方法，并把这个任务需要完成的具体工作在deal方法中实现。
 *  线程池中的线程之所以被创建，就是为了执行各种各样的任务，为了方便线程对任务的处理，
 *  需要用一个Task抽象类来保证任务的具体工作同一放在deal方法中来完成。
 */
public abstract class Task {
	public enum State{
		/*新建*/NEW,
		/*运行*/RUNNING,
		/*结束*/FINISHED
	}
	/*
	 * 任务状态
	 */
	private State state=State.NEW;
	
	public State getState() {
		return state;
	}
	public void setState(State state) {
		this.state = state;
	}
	public abstract void deal();
}

  TaskQueue(任务队列)

/**
 * 任务队列，在同一时刻，可能有很多任务需要执行，而程序在同一时刻只能执行一定数量的任务，
 * 当需要执行的任务数超过了程序所能承受的任务数后怎么办，这就有了先执行那些任务，后执行
 * 那些任务的规则。TaskQueue类就是这些规则中的一种，采用了FIFO(先进先出)，也就是按照
 * 任务到达的先后顺序执行。
 */
public class TaskQueue {
	
	private List<Task> queue=new LinkedList<Task>();
	
	//添加一项任务
	public synchronized void addTask(Task task){
		if(task!=null){
			queue.add(task);
		}
	}
	//完成任务后将它从任务队列中删除
	public synchronized void finishTask(Task task){
		if(task!=null){
			task.setState(Task.State.FINISHED);
			queue.remove(task);
		}
	}
	//取得一项待执行任务
	public synchronized Task getTask(){
		Iterator<Task> iterator=queue.iterator();
		Task task;
		while(iterator.hasNext()){
			task=iterator.next();
			//寻找一个新建的任务
			if(Task.State.NEW.equals(task.getState())){
				//把任务状态置为运行中
				task.setState(Task.State.RUNNING);
				return task;
			}
		}
		return null;
	}
}

 TaskThread(执行任务的线程)

/*
 * TaskThread是线程池中默认的线程处理程序，
 * 是执行任务的线程，专门用于执行任务队列中的待执行任务
 * 不断地从任务队列中取出任务，然后执行
 */
public class TaskThread extends Thread {
	//该线程所属的线程池
	private ThreadPoolService service;
	public TaskThread(ThreadPoolService tps){
		this.service=tps;
	}
	public void run(){
		//在线程池运行的状态下执行 任务队列中的任务
		while(service.isRunning()){
			TaskQueue queue=service.getTaskQueue();
			Task task=queue.getTask();
			if(task!=null){
				task.deal();
			}
			queue.finishTask(task);
		}
	}
}

 ThreadPoolService（线程池服务类）

import java.util.ArrayList;
import java.util.List;
/**
 * 在创建的时候就创建了几个线程对象，但是线程并没有启动运行。
 * 在调用了start方法启动线程池服务之后，它们才真正运行。
 * stop方法可以停止线程池服务，同时通知池中所有线程的运行。
 * runTask(Task task)将一个新的待执行任务交与线程池来运行。
 */

public class ThreadPoolService {
	public enum Status{
		NEW,
		RUNNING,
		TERMINATED
	}
	 //线程数
	public static final int THREAD_COUNT=5;
   //线程池状态
	private Status status=Status.NEW;
	private TaskQueue queue=new TaskQueue();
	private List<Thread> threads=new ArrayList<Thread>();
	public ThreadPoolService(){
		for(int i=0;i<THREAD_COUNT;i++){
			Thread t=new TaskThread(this);
			threads.add(t);
		}
	}
	//启动服务
	public void start(){
		this.status=Status.RUNNING;
		for(int i=0;i<THREAD_COUNT;i++){
			threads.get(i).start();
		}
	}	
	//停止服务
	public void stop(){
		this.status=Status.TERMINATED;
	}
	//是否在运行
	public boolean isRunning(){
		return status==Status.RUNNING;
	}
	//执行任务
	public void runTask(Task task){
		queue.addTask(task);
	}
	
	protected TaskQueue getTaskQueue(){
		return queue;
	}
}

       完成了上面四个类，我们就实现了一个简单的线程池。现在我们就可以使用它了，下面的代码做了一个简单的示例。

class SimpleTask extends Task {

	private int i;
	public SimpleTask(int i) {
		super();
		this.i = i;
	}

	@Override
	public void deal() {
		System.out.println(Thread.currentThread().getName()+"-----"+"我是SimpleTask : " + i);
	}
}

public class ThreadPoolTest {
	public static void main(String[] args) throws InterruptedException {
		ThreadPoolService service=new ThreadPoolService();
		service.start();
		for(int i=0;i<20;i++){
			service.runTask(new SimpleTask(i));
		}
		Thread.sleep(5000);
		service.stop();
	}
}

       当然，我们实现的是最简单的，这里只是为了演示线程池的实现原理。在实际应用中，根据情况的不同，可以做很多优化。比如：

       调整任务队列的规则，给任务设置优先级，级别高的任务优先执行。

       动态维护线程池，当待执行任务数量较多时，增加线程的数量，加快任务的执行速度；当任务较少时，回收一部分长期闲置的线程，减少对系统资源的消耗。

       事实上JAVA5.0及以上版本已经为我们提供了线程池功能，无需再重新实现。这些类位于java.util.concurrent包中。

         Executors类提供了一组创建线程池对象的方法，常用的有以下几个：

Executors.newCachedThreadPool();
Executors.newFixedThreadPool(int nThreads);
Executors.newSingleThreadExecutor();

        newCachedThreadPool方法创建一个动态的线程池，其中线程的数量会根据实际需要来创建和回收，适合于执行大量短期任务的情况；newFixedThreadPool(int nThreads)方法创建一个包含固定数量线程对象的线程池，nThreads代表要创建的线程数，如果某个线程在运行的过程中因为异常而终止了，那么一个新的线程会被创建和启动来代替它；而newSingleThreadExecutor()方法则只在线程池中创建一个线程，来执行所有的任务

          这三个方法都返回了一个ExecutorService类型的对象。实际上，ExecutorService是一个接口，它的submit()方法负责接收任务并交与线程池中的线程去运行。submit()方法能够接收Callable和Runnable两种类型的对象。它们的用法和区别如下：

1、Runnable接口：继承Runnable接口的类要实现它的run()方法，并将执行任务的代码放入其中，run()方法没有返回值。适合于只作某种操作，不关心运行结果的情况。

2、Callable接口：继承Callable接口的类要实现它的call()方法，并将执行任务的代码放入其中，call()将任务的执行结果作为返回值。适合于执行某种操作之后，需要知道执行结果的情况。

       无论是接收Runnable型参数，还是接收Callable型参数的submit()方法，都会返回一个Future(也是一个接口)类型的对象。该对象中包含了任务的执行情况以及结果。调用Future的boolean isDone方法可以知道任务是否执行完毕；调用get()方法可以获得任务执行后的返回结果，如果此时任务还没有执行完，get()方法会保持等待，直到相应的任务执行完毕后，才会将结果返回。

我们用一个例子来演示JAVA5.0中的线程池的使用：

public class ExecutorTest {

	/**
	 * @param args
	 * @throws ExecutionException 
	 * @throws InterruptedException 
	 */
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		// TODO Auto-generated method stub
		ExecutorService es=Executors.newCachedThreadPool();
		//提交任务
		Future fr=es.submit(new RunnableTest());
		//提交任务
		Future fc=es.submit(new CallableTest());
		//取得返回值并输出
		System.out.println(fc.get());
		
		//检查任务是否执行完毕
		if(fr.isDone()){
			System.out.println("执行完毕-RunnableTest.run()");
		}else{
			System.out.println("未执行完毕-RunnableTest.run()");
		}
		//检查任务是否执行完毕
		if(fc.isDone()){
			System.out.println("执行完毕-CallableTest.run()");
		}else{
			System.out.println("未执行完毕-CallableTest.run()");
		}
		//停止线程池服务
		es.shutdown();
	}
}

class RunnableTest implements Runnable{
	public void run() {
		System.out.println("已经执行-RunnableTest.run()");
	}
}
class CallableTest implements Callable{

	public Object call() throws Exception {
		// TODO Auto-generated method stub
		System.out.println("已经执行-CallableTest.call()");
		return "返回值-CallableTest.call()";
	}
}

      运行结果：

已经执行-RunnableTest.run()
已经执行-CallableTest.call()
返回值-CallableTest.call()
执行完毕-RunnableTest.run()
执行完毕-CallableTest.run()
       使用完线程池之后，需要调用它的shutdown()方法停止服务，否则其中的所有线程都会保持运行，程序不会退出。

评论

1 楼 cy297179121 2015-11-13  

ThreadPoolService类中的queue、threads之间什么关系？ThreadPoolTest测试类中没有体现出二者的关系啊？？？求指教