pipIrr-SV.git - Gitblit

New file
			@@ -0,0 +1,439 @@
			package com.dy.common.threadPool;

			import java.util.*;

			import org.apache.logging.log4j.*;

			import com.dy.common.threadPool.ThreadPool.Job;

			public class ThreadPoolImp {

			/**
			* 线程池实现
			* @author Administrator
			*
			*/
			public class MyThreadPool implements ThreadPool.Pool{

			/**
			* 线程池唯一实例
			*/
			public MyThreadPool myPool ;

			/**
			* 线程池名称；
			*/
			private String poolName ;
			/**
			* 空闲线程集合
			*/
			private List<MyThread> freeThreads;

			/**
			* 工作中的线程集合
			*/
			private List<MyThread> busiThreads;

			/**
			* 线程池最大线程数 , 若maxNum <= 0，则maxNum = 1 。
			*/
			private int maxNum;

			/**
			* 最小线程数，若minNum > maxNum，则minNum = maxNum 。
			*/
			private int minNum;

			/**
			* 当前线程池中的线程数。
			*/
			private int currNum;

			/**
			* 空闲线程超时的时长(秒)，超过这个时间，就清除多余线程，
			* 使空闲线程最多是minNum个
			*/
			private long freeTimeout ;

			/**
			* 忙碌线程超时的时长(秒)，超过这个时间，就认为是崩溃线程，将被清除。
			*/
			private long busyTimeout ;

			/**
			* 同步对象
			*/
			private Object synObj = new Object();

			/**
			* 内部启动的空闲和忙碌超时线程的线程
			*/
			private Thread monitorThread ;

			/**
			* 日志
			*/
			private static final Logger log = LogManager.getLogger(MonitorThread.class) ;

			/**
			* 线程池构造方法
			* @param poolName 线程池和线程名称
			* @param maxNum 线程池最大线程数，若为-1，不受限制
			* @param minNum 线程池最小线程数，或初始线程数
			* @param freeTimeout 空闲线程超时时长(秒)
			* @param busyTimeout 忙碌线程超时时长(秒)，若为-1，不受限制
			*/
			public MyThreadPool(
			String poolName ,
			Integer maxNum ,
			Integer minNum ,
			Long freeTimeout ,
			Long busyTimeout) {
			if(poolName == null){
			poolName="线程池" ;
			}
			this.poolName = poolName ;

			if(maxNum == null \|\| maxNum.intValue() < 0){
			maxNum = 65535 ;
			}
			if(minNum == null \|\| minNum.intValue() < 0){
			minNum = 0 ;
			}
			if(minNum > maxNum){
			minNum = maxNum ;
			}
			this.maxNum = maxNum ;
			this.minNum = minNum ;
			this.currNum = 0;

			if(freeTimeout == null \|\| freeTimeout.longValue() <= 0){
			freeTimeout = 60000L ;
			}
			this.freeTimeout = freeTimeout ;

			if(busyTimeout == null \|\| busyTimeout.longValue() <= 0){
			this.busyTimeout = -1L ;
			}else{
			this.busyTimeout = busyTimeout ;
			}
			if(maxNum != 0){
			this.busiThreads = new ArrayList<>();
			this.freeThreads = new ArrayList<>();
			//最小化线程池
			for (int i = 0; i < this.minNum ; i++) {
			MyThread t = new MyThread(this);
			t.start();
			this.freeThreads.add(t);
			this.currNum++;
			}
			this.monitorThread = new MonitorThread(this) ;
			this.monitorThread.start() ;
			}
			}
			/**
			* 线程池中线程个数
			* @return
			*/
			@Override
			public Integer size() {
			return currNum ;
			}
			@Override
			public Integer maxThread() {
			return maxNum ;
			}
			@Override
			public Integer minThread() {
			return minNum ;
			}

			/**
			* 把所要执行的工作对象实例放入线程池中
			* @param job ThreadJob 工作对象实例
			* @throws Exception
			*/
			@Override
			public void putJob(Job job) throws Exception {
			if(this.busiThreads == null \|\| this.freeThreads == null){
			throw new Exception("线程池未启动") ;
			}
			synchronized(this.synObj) {
			//log.debug("工作任务分配到线程池中。") ;
			MyThread t = null ;
			if (this.freeThreads.size() == 0) {
			//当前没有空闲线程
			if (this.maxNum == -1 \|\| this.currNum < this.maxNum) {
			//当前线程数未达到最大值 , 增加新的线程
			t = new MyThread(this);
			t.start();
			this.currNum++;
			} else {
			//当前线程达到最大数了，等待回归线程
			while (freeThreads.size() == 0) {
			//如果没有空闲的线程,等待工作线程工作完成回来
			try {
			//log.warn("线程池(" + this.poolName + ")中线程数达到上限，新工作任务等待释放线程!");
			synObj.wait();
			} catch (Exception e) {
			log.error("'" + this.poolName + "'线程池中线程等待释放线时发生等待异常!", e);
			}
			t = (MyThread) freeThreads.get(0);
			if (t != null) {
			//说明得到了释放回来的线程
			freeThreads.remove(0);
			break;
			}else{
			//说明没有得到释放回来的线程，可以其他线程
			continue;
			}
			}//end while
			}//end else
			}//end if(freeThreads.size() == 0)
			else {
			t = (MyThread) freeThreads.get(0);
			freeThreads.remove(0);
			}
			busiThreads.add(t);
			t.putJob(job);
			}
			}

			/**
			* 线程工作完成，从busiThreads回归freeThreads
			*/
			protected void freeThread(MyThread t) throws Exception {
			if(this.busiThreads == null \|\| this.freeThreads == null){
			throw new Exception("线程池未启动") ;
			}
			synchronized (synObj) {
			busiThreads.remove(t);
			freeThreads.add(t);
			synObj.notify();
			}
			}

			/**
			* 监控超时线程的线程
			* @author Administrator
			*
			*/
			protected class MonitorThread extends Thread {
			private MyThreadPool pool ;

			private MyThread t ;
			private Iterator<?> it ;

			/**
			*
			* @param pool 池
			*/
			public MonitorThread(MyThreadPool pool){
			this.pool = pool ;
			}
			/**
			*
			*/
			@SuppressWarnings("finally")
			public void run(){
			long interval = pool.freeTimeout ;
			if(pool.busyTimeout > 0 && pool.busyTimeout < pool.freeTimeout){
			interval = pool.busyTimeout ;
			}
			boolean isException = false ;
			while(true){
			t = null ;
			it = null ;
			try{
			MonitorThread.sleep(interval) ;
			}catch(Exception e){
			isException = true ;
			}finally{
			if(isException){
			isException = false ;
			continue ;
			}
			}
			try{
			synchronized (pool.synObj) {
			if(pool.freeThreads.size() > pool.minNum){
			//如果空闲线程大于最小线程数，则清理空闲线程
			int num = pool.freeThreads.size() - pool.minNum ;
			int count = 0 ;
			it = pool.freeThreads.iterator() ;
			while(it.hasNext()){
			if(count == num) {
			break ;
			}
			count ++ ;
			t = (MyThread)it.next() ;
			if((System.currentTimeMillis() - t.time) >= pool.freeTimeout){
			it.remove() ;
			pool.currNum-- ;
			//log.debug("线程池(" + pool.poolName + ")中清除了一个超时空闲线程!");
			t.destroy() ;
			t = null ;
			}
			}
			}//end if

			if(pool.busyTimeout != -1 && pool.busyTimeout > 0){
			it = pool.busiThreads.iterator() ;
			while(it.hasNext()){
			t = (MyThread)it.next() ;
			if((System.currentTimeMillis() - t.time) >= pool.busyTimeout){
			it.remove() ;
			pool.currNum-- ;
			log.error("线程池(" + pool.poolName + ")中清除了一个超时崩溃(忙碌)线程!");
			t.destroy() ;
			t = null ;
			}
			}
			}
			}//end synchronized (pool.synObj)
			}catch(Exception e){
			e.printStackTrace();
			}finally{
			continue ;
			}
			}//end while(true)
			}
			}


			}

			/**
			* 池中线程实现
			* @author Administrator
			*
			*/
			public class MyThread extends Thread{

			/**
			* 标示线程是活的
			*/
			private boolean living = true ;

			/**
			* 线程忙碌或空闲记时器
			*/
			protected long time ;

			/**
			* 当前线程所处的线程池
			*/
			private MyThreadPool pool ;
			/**
			* 线程具体工作的回调类
			*/
			private Job job ;

			/**
			* 指示线程可以工作
			*/
			private Boolean canJob ;

			private Logger log = LogManager.getLogger(MyThread.class.getName()) ;

			protected MyThread(MyThreadPool pool) {
			super();
			this.pool = pool ;
			this.time = 0 ;
			this.canJob = false ;
			}

			/**
			* 设置线程工作对象
			* @param job 工作
			*/
			protected void putJob(Job job) throws Exception {
			if(job == null){
			this.job = new Job(){
			public void execute(){
			}
			public void destroy(){
			}
			public boolean isDestroy(){
			return false ;
			}
			};
			}
			synchronized (this) {
			this.job = job ;
			this.canJob = true;
			//忙碌记时开始
			this.time = System.currentTimeMillis() ;
			this.notify();
			}
			}

			/**
			*
			*/
			@SuppressWarnings("finally")
			@Override
			public void run(){
			while (living){
			synchronized (this){
			while(!canJob){
			//当不能工作时
			try{
			this.wait();
			}catch (Exception e) {
			log.error("线程池(" + pool.poolName + ")的工作线程等待可以工作的信号时发生等待异常:\n" + e.getMessage(), e);
			this.canJob = false ;
			continue;
			}
			}
			///////////////////////
			//被唤醒，有新工作了
			try{
			if(this.job != null){
			this.job.execute() ;
			}
			}catch (Exception ee) {
			log.error("线程池(" + pool.poolName + ")的工作线程在执行工作时发生异常:\n" + ee.getMessage(), ee);
			//ee.printStackTrace() ;
			}finally {
			this.canJob = false ;
			this.job = null ;
			if(living){
			//线程未被销毁
			this.free() ;
			}
			continue;
			}
			}// end synchronized(this)
			}// end while(living)
			}

			public void free(){
			try{
			//使本线程回归空闲线程池
			pool.freeThread(this);
			//空闲开始记时
			this.time = System.currentTimeMillis() ;
			// 没有可做的了
			this.canJob = false;
			log.debug("线程池(" + this.pool.poolName + ")中的线程回归空闲集合。");
			}catch (Exception e){
			log.error("线程池(" + pool.poolName + ")的工作线程释放回归时发生异常:\n" + e.getMessage(), e);
			e.printStackTrace();
			}

			}

			/**
			* 销毁工作线程
			*/
			public void destroy() {
			//线程销毁前，首先把其所持有的工作销毁，否则线程停不下来，也销毁不了
			if(this.job != null){
			this.job.destroy();
			}
			this.living = false ;//使线程从run方法的长久while(living)中退出来，从线程自动销毁
			this.job = null ;
			}
			}
			}