线程池

1、一个阻塞队列，用来存放执行的任务：

package com.baidu.test;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @功能：阻塞队列
 * @作者: smile
 * @时间：2013-4-24 下午12:36:23
 * @版本：1.0
 */
public class BlockQueue<E> {
	
	/**	队列存放的元素  **/
	private E[] items;
	/** 插入元素开始位 **/
	private int putIndex;
	/**	移除元素结束位**/
	private int takeIndex;
	/**	队列中元素数量**/
	private int count;
	
	/**	操作锁**/
	private ReentrantLock lock;
	/**	已满条件变量	**/
	private Condition full;
	/**	已空条件变量	**/
	private Condition empty;
	
	public BlockQueue(int capacity){
		if(capacity<=0){
			throw new IllegalArgumentException("参数必须大于0");
		}
		items = (E[])new Object[capacity];
		full = lock.newCondition();
		empty = lock.newCondition();
	}
	
	/** 插入一个元素	**/
	public void put(E e){
		if(e == null){
			throw new NullPointerException();
		}
		lock.lock();
		try {
			while(count == items.length){
				//队列已满  插入线程加入等待池中
				full.await();
			}
			//插入 
			items[putIndex] = e;
			++count;
			//如果已经插入到队列最后一个索引位 则下一次插入要从头开始
			putIndex = (++putIndex == items.length) ? 0 : putIndex;
			//随机唤醒一个删除线程
			empty.signal();
		}catch (Exception ex) {
			ex.printStackTrace();
		}finally{
			lock.unlock();
		}
	}
	
	/**	移除一个元素	**/
	public E take(){
		lock.lock();
		try {
			while(count == 0){
				//队列已空  移除线程等待
				empty.await();
			}
			E e = items[takeIndex];
			items[takeIndex] = null;
			--count;
			//如果已经删除到队列最后一个索引位 则下一次删除要从头开始
			takeIndex = (++takeIndex == items.length) ? 0 : takeIndex;
			//随机唤醒一个插入线程
			full.signal();
			return e;
		} catch (Exception e) {
			e.printStackTrace();
		}finally{
			lock.unlock();
		}
		return null;
	}
	
	//是否已满
	public boolean isFull(){
		lock.lock();
		try {
			return (items.length == count);
		}finally{
			lock.unlock();
		}
	}
}

 2、一个超额任务拒绝策略，为简便这里没封装成接口：

package com.baidu.test;

/**
 * @功能：超额任务处理策略
 * @作者: smile
 * @时间：2013-4-24 下午1:11:25
 * @版本：1.0
 */
public class RejectHandler {
		
	//处理策略
	public static void handle(Runnable run){
		if(run != null){
			//直接执行
			run.run();
		}
	}
}

 3、一个线程工厂，用来创建线程：

package com.baidu.test;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @功能：线程工厂
 * @作者: smile
 * @时间：2013-4-24 下午1:06:29
 * @版本：1.0
 */
public class ThreadFactory {

	//线程数量原子量
	private static AtomicInteger number = new AtomicInteger(1);
	//线程名前缀
	private String pre;
	
	public Thread newThread(Runnable run){
		pre = "线程"+number.getAndIncrement();
		Thread t = new Thread(run, pre);
		return t;
	}
}

 4、线程池：

package com.baidu.test;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @功能：线程池
 * @作者: smile
 * @时间：2013-4-24 下午1:19:27
 * @版本：1.0
 */
public class ThreadPool {
	
	/**	线程池状态 **/
	private volatile int runState;
	/**	运行 **/
	static final int RUNNING = 0;
	/**	停止 **/
	static final int STOP = 1;
	/**	任务队列 **/
	private final BlockQueue<Runnable> taskQueue;
	/**	操作锁 **/
	private final ReentrantLock mainLock = new ReentrantLock();
	
	/** 工作任务集合 后续比如shutdown会用到**/
	private Set<Worker> work = new HashSet<Worker>();
	
	/**	基本线程数 **/
	private volatile int corePoolSize;
	/** 最大线程数 **/
	private volatile int maxPoolSize;
	/** 最大空闲时间  操过比如线程数大于基本线程数则释放该线程**/
	private volatile long maxFreeTime;
	/**	真实线程数 **/
	private volatile int poolSize;
	/** 拒绝策略 可使用接口抽象出一个拒绝策略接口  并提供不同拒绝策略**/
	private RejectHandler reject;
	/** 线程生成工厂  可抽象出接口 **/
	private ThreadFactory facotry;
	
	//线程池默认构造函数  简单写了
	public ThreadPool(){
		taskQueue = new BlockQueue<Runnable>(20);
		this.corePoolSize = 10;
		this.maxPoolSize = 20;
		this.maxFreeTime = 10000;
		this.reject = new RejectHandler();
		this.facotry = new ThreadFactory();
	}
	
	/** 执行任务 **/
	public void execute(Runnable run){
		if(run == null){
			throw new NullPointerException();
		}
		//简单处理了  
		//本来是少于基本线程则创建 多于基本线程小于最大线程 则放入队列中  如果队列满 则创建线程知道最大线程数  如果还不够则采用拒绝策略
		if(poolSize <= corePoolSize){
			//需要新增工作线程
			mainLock.lock();
			Thread t = null;
			try {
				//使用工作任务包装任务线程 
				Worker w = new Worker(run);
				work.add(w);
				t = facotry.newThread(w);
				//指定任务的相应线程
				w.thread = t;
				poolSize++;
			} catch (Exception e) {
				e.printStackTrace();
			}finally{
				mainLock.unlock();
			}
			//启动工作线程 而不是任务线程
			t.start();
		}else if(poolSize<=maxPoolSize){
			//加入阻塞队列 
			if(taskQueue.isFull()){
				taskQueue.put(run);
			}
			//拒绝策略处理这个任务
			reject.handle(run);
		}
	}
	
	//返回队列下一个任务
	private Runnable getTask(){
		return taskQueue.take();
	}
	
	//当前工作任务已经执行完 队列中无任务
	public void workDone(Runnable run){
		mainLock.lock();
		try {
			//把当前工作任务从work中移除
			work.remove(run);
			poolSize--;
		} catch (Exception e) {
			e.printStackTrace();
		}finally{
			mainLock.unlock();
		}
	}
	
	/**
	 * @功能：工作任务 包装真实任务  这里边主要是使用一个工作线程执行多个真实任务
	 */
	class Worker implements Runnable{

		public Runnable firstTask;
		//工作线程
		public Thread thread;
		
		public Worker(Runnable firstTask){
			this.firstTask = firstTask;
		}
		
		//执行第一个线程，一次执行到队列中所有线程直到队列为空
		public void run() {
			try {
                while (firstTask != null || (firstTask = getTask())!=null) {
                	if(runState == STOP){
						//如果线程池停止  则停止当前任务线程执行
                		thread.interrupted();
					}
                	//执行任务
					firstTask.run();
                	firstTask = null;
                }
            } finally {
                //队列任务全部执行完  从工作队列中
            	workDone(this);
            }
		}
	}
}

 5、基本的流程是这样的：

1、初始化线程池，初始化基本线程数，最大线程数，线程最大空闲时间，存放任务的队列，创建线程的工厂，超额任务拒绝策略。
2、通过execute加入线程，如果线程数小于基本线程数，则通过线程工厂创建线程，并通过工作任务封装下原来的任务，工作任务run方法中会执行目标任务，并且循环查询队列中的任务依次执行，知道任务全部执行完。如果线程池中线程已经大于基本线程数且线程池是运行的，则把任务加入到任务队列，如果队列已满，则只能新建线程来执行知道线程数到达最大线程数，如果到达了还有任务过来，那就只有通过任务拒绝策略处理了。
3、工作任务线程这里就是一个线程可以处理多个任务。

 6、线程池核心方法：

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
            if (runState == RUNNING && workQueue.offer(command)) {
                if (runState != RUNNING || poolSize == 0)
                    ensureQueuedTaskHandled(command);
            }
            else if (!addIfUnderMaximumPoolSize(command))
                reject(command); // is shutdown or saturated
        }
    }