线程池

package threadpool;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Collection;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;

public class ThreadPoolManager {
	private static Logger log = Logger.getLogger(ThreadPoolManager.class);
	
	private static ThreadPoolManager instance = null;
	private int poolSize = 10;
	private final TaskThread[] taskThread;
	private final Queue<Runnable> workQueue;
	
	private ThreadPoolManager() {
		workQueue = new LinkedBlockingQueue<Runnable>();//创建工作任务队列
		taskThread = new TaskThread[poolSize];//初始化线程池
		for(int i=0;i<poolSize;i++){
			taskThread[i] = new TaskThread();
			taskThread[i].start();
			log.info("start one Thread["+i+"] ID:"+taskThread[i].hashCode());
		}
	}
	/**
	 * 获取线程池对象
	 * @return
	 */
	public synchronized static ThreadPoolManager getSingleInstance(){
		if(ThreadPoolManager.instance == null){
			ThreadPoolManager.instance = new ThreadPoolManager();
			log.info("get one ThreadPoolManager new instance ID:"+instance.hashCode());
		}
		return ThreadPoolManager.instance;
	}
	/**
	 * 添加单个任务
	 * @param r
	 * @throws FileNotFoundException
	 * @throws IOException
	 * @throws ClassNotFoundException
	 */
	public void executeTask(Runnable r){
		synchronized(workQueue){
			int size = workQueue.size();
			if(size > 2000000){//任务数操作200w
				workQueue.notifyAll();
				try {
					workQueue.wait(20);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				workQueue.add(r);
			}else{
				workQueue.add(r);
				workQueue.notifyAll();
			}
		}
	}
	
	/**
	 * 批量添加任务数组
	 * @param rs
	 * @throws FileNotFoundException
	 * @throws IOException
	 * @throws ClassNotFoundException
	 */
	public void batchExceuteTask(Runnable[] rs){
		for(Runnable r : rs){
			executeTask(r);
		}
	}
	/**
	 * 批量添加任务集合
	 * @param rs
	 * @throws FileNotFoundException
	 * @throws IOException
	 * @throws ClassNotFoundException
	 */
	public void batchExceuteTask(Collection<Runnable> rs){
		for(Runnable r : rs){
			executeTask(r);
		}
	}
	/**
	 * 立即停止正在执行的线程并销毁资源
	 */
	public synchronized void destory(){
		for(int i=0;i<taskThread.length;i++){
			if(taskThread[i] != null){
				log.info("Thread ["+i+"] ID:"+taskThread[i].hashCode()+" is close!");
				taskThread[i].threadStop();
			}
			taskThread[i] = null;
		}
		synchronized(workQueue){
			workQueue.clear();
			workQueue.notifyAll();
		}
		ThreadPoolManager.instance = null;
	}
	/**
	 * 等待任务结束在停止线程并销毁资源
	 */
	public void delayDestory(){
		try {
			while(!workQueue.isEmpty())
			Thread.sleep(1000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		destory();
	}
	/**
	 * 线程池中的实体线程类
	 * @author pc001
	 */
	private class TaskThread extends Thread{
		private boolean isRunning = true;
		
		public void run() {
			while(isRunning){
				Runnable r = null;
				synchronized(workQueue){
					if(workQueue.isEmpty()){
						try {
							workQueue.wait();
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
				}
				r = workQueue.poll();
				if(r != null)
				r.run();
			}
		}
		/**
		 * 线程开关
		 */
		public void threadStop(){
			this.isRunning = false;
		}
	}
	
	public static void main(String[] args){
		ThreadPoolManager manager = ThreadPoolManager.getSingleInstance();
		LinkedList<Runnable> list = new LinkedList<Runnable>();
		for(int i=0;i<10000;i++){
			list.add(new TaskRun());
		}
		manager.batchExceuteTask(list);
		manager.delayDestory();
	}
}

学习了一下线程池的思想，自己写了一个线程池实现类，解决的普通线程池无法承压快速提交的任务造成的内存溢出错误，经过自己测试可以无限添加任务数量，而且自己实现了两种销毁线程池对象的方法，自己感觉有些bug如果有哪位高人完善此线程请务必贴出来，供大家学习。毕竟我一个人的能力有限。呵呵！

改进：

package thread;

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

public class ThreadPool {
	private final int 			POOL_SIZE;
	private final Thread[] 		THREAD_POOL;
	
	private final Runnable[] 	TASK_LIST = new Runnable[1000];
	private int count  = 0;
	private int rIndex = 0;
	private int wIndex = 0;
	
	private final ReentrantLock LOCK  = new ReentrantLock();
	private final Condition 	FULL  = LOCK.newCondition();
	private final Condition 	EMPTY = LOCK.newCondition();
	
	private volatile boolean running = true;
	
	public ThreadPool() {
		this( 10 );
	}
	
	public ThreadPool( int capacity ){
		this.POOL_SIZE 	 = capacity;
		this.THREAD_POOL = new Thread[ this.POOL_SIZE ];
		initThreadPool();
	}
	
	private void initThreadPool(){
		for( int i = 0 ; i < this.THREAD_POOL.length ; i++ ){
			this.THREAD_POOL[i] = new Thread( new Runner() );
			this.THREAD_POOL[i].start();
		}
	}
	
	private boolean isEmpty(){
		return this.count == 0;
	}
	
	private boolean isFull(){
		return this.count == TASK_LIST.length;
	}
	
	private void push( Runnable task ){
		LOCK.lock();
		try{
			if( !running ) return;
			if( this.isFull() ){
				try {
					FULL.await();
				} catch (InterruptedException e) {
					FULL.signal();
				}
			}
			if( wIndex > TASK_LIST.length-1 )
				wIndex = 0;
			count++;
			TASK_LIST[wIndex++] = task;
			EMPTY.signal();
		}finally{
			LOCK.unlock();
		}
	}
	
	private Runnable poll(){
		Runnable task = null;
		LOCK.lock();
		try{
			if( isEmpty() ){
				try {
					EMPTY.await();
				} catch (InterruptedException e) {
					EMPTY.signal();
				}
			}
			if( rIndex > TASK_LIST.length-1 )
				rIndex = 0;
			count--;
			task = TASK_LIST[rIndex++];
			FULL.signal();
		}finally{
			LOCK.unlock();
		}
		return task;
	}
	
	public void execute( Runnable task ){
		push(task);
	}
	
	public void shutdown(){
		LOCK.lock();
		try{
			running = false;
			FULL.signal();
			EMPTY.signal();
			for( Thread thread : this.THREAD_POOL ){
				if( thread != Thread.currentThread() )
					thread.interrupt();
			}
		}finally{
			LOCK.unlock();
		}
	}
	
	private class Runner implements Runnable{
		public void run() {
			while( running ){
				Runnable task = poll();
				if( task != null ){
					try {
						task.run();
					} catch (Exception e) {
					}
				}
			}
		}
	}
	
	public static void main(String[] args) {
		final ThreadPool pool = new ThreadPool();
		final Thread t = new Thread( new Runnable(){
			public void run() {
				for( int i =0;i<1000000;i++ ){
					pool.execute(new Runnable(){
						@Override
						public void run() {
							System.out.println("test");
						}
					});
				}
			}
		});
		t.start();
		new Thread( new Runnable(){
			public void run() {
				try {
					Thread.sleep(3000);
				} catch (InterruptedException e) {
				}
				pool.shutdown();
//				t.interrupt();
			}
		}).start();
	}
}