- 最后登录
- 2013-6-5
- 注册时间
- 2011-7-16
- 阅读权限
- 90
- 积分
- 6011
 
- 纳金币
- 5996
- 精华
- 1
|
import java.util.Vector;
import java.net.*;
import java.io.*;
public class ThreadPool {
public static final int MAX_THREADS = 100;
public static final int MAX_SPARE_THREADS = 50;
public static final int MIN_SPARE_THREADS = 10;
public static final int WORK_WAIT_TIMEOUT = 60 * 1000;
protected Vector pool; //存放空闲线程
protected MonitorRunnable monitor; //A monitor thread that monitors the pool for idel threads.
protected int maxThreads; //Max number of threads that you can open in the pool.
protected int minSpareThreads; //Min number of idel threads that you can leave in the pool.
protected int maxSpareThreads; //Max number of idel threads that you can leave in the pool.
protected int currentThreadCount; //Number of threads in the pool.
protected int currentThreadsBusy; //Number of busy threads in the pool.
protected boolean stopThePool; //Flag that the pool should terminate all the threads and stop.
/**
* Construct
*/
public ThreadPool() {
maxThreads = MAX_THREADS;
maxSpareThreads = MAX_SPARE_THREADS;
minSpareThreads = MIN_SPARE_THREADS;
currentThreadCount = 0;
currentThreadsBusy = 0;
stopThePool = false;
}
/**
* 启动线程池
*/
public synchronized void start() {
adjustLimits(); //调整最大和最小线程数及最大和最小多余线程数.
openThreads(minSpareThreads); //打开初始线程
monitor = new MonitorRunnable(this); //Runnable对象实例 //A monitor thread that monitors the pool for idel threads.
}
public void setMaxThreads(int maxThreads) {
this.maxThreads = maxThreads;
}
public int getMaxThreads() {
return maxThreads;
}
public void setMinSpareThreads(int minSpareThreads) {
this.minSpareThreads = minSpareThreads;
}
public int getMinSpareThreads() {
return minSpareThreads;
}
public void setMaxSpareThreads(int maxSpareThreads) {
this.maxSpareThreads = maxSpareThreads;
}
public int getMaxSpareThreads() {
return maxSpareThreads;
}
/**
* 线程池管理方法.
* 当空闲队列线程中没有空闲线程时,则增加处理(空闲)线程数量.
* 如果线程数量已达到最大线程数,则新的连接进行等待.
* 当请求到来,且有空闲线程时调用处理线程进行具体业务处理.
* @param r ThreadPoolRunnable
*/
public void runIt(Socket cs) { //r 为task //有任务进入时调用
if (null == cs) {
throw new NullPointerException();
}
if (0 == currentThreadCount || stopThePool) {
throw new IllegalStateException();
}
ControlRunnable c = null; //任务处理实例.
synchronized (this) {
if (currentThreadsBusy == currentThreadCount) { //如果工作线程和当前线程数相等,说明没有空闲线程.
if (currentThreadCount < maxThreads) { //如果当前线程数还没有达到最大线程数.
int toOpen = currentThreadCount + minSpareThreads; //再增加minSpareThreads个线程量.
openThreads(toOpen); //打开线程新增空闲线程. //currentThreadCount数量增加
}
else { //如果当前数量达到了最大线程数.
while (currentThreadsBusy == currentThreadCount) { //当工作线程和当前线程数相等,说明没有空闲线程.
try {
this.wait(); //连接线程进行等待.
}
catch (InterruptedException e) {
}
if (0 == currentThreadCount || stopThePool) {
throw new IllegalStateException();
}
}
}
}
c = (ControlRunnable) pool.lastElement(); //在有空闲线程的情况下,从空闲线程队列中取出最后一个线程.
pool.removeElement(c); //从空闲队列中删除最后一个线程,用于处理其他事件.
currentThreadsBusy++; //对处理事件的线程数加1
}
System.out.println("系统调用一个Sokcet线程");
c.runIt(cs); //调用具体业务方法,告诉其有数据请求要处理,唤醒等待中的线程.
}
/**
* 关闭线程池
*/
public synchronized void shutdown() {
if (!stopThePool) { //如果线程池没有关闭,(线程池关闭标识为假)
stopThePool = true;
monitor.terminate(); //关闭监视线程
monitor = null;
for (int i = 0; i < (currentThreadCount - currentThreadsBusy); i++) { //关闭空闲线程队列
try {
( (ControlRunnable) (pool.elementAt(i))).terminate();
}
catch (Throwable t) {
}
}
currentThreadsBusy = currentThreadCount = 0;
pool = null;
notifyAll(); //唤醒所有在等待的线程.
}
}
/**
* 当线程大于最大多余线程时关闭多余的线程.
*/
protected synchronized void checkSpareControllers() {
if (stopThePool) { //如果连接池没有关闭.
return;
}
if ( (currentThreadCount - currentThreadsBusy) > maxSpareThreads) { //如果空闲的线程数大于多余的最大线程数量.
int toFree = currentThreadCount - currentThreadsBusy - maxSpareThreads; //得出多余的线程数量
for (int i = 0; i < toFree; i++) { //关闭删除空闲线程,从Vector中删除
ControlRunnable c = (ControlRunnable) pool.firstElement();
pool.removeElement(c);
c.terminate(); //让删除的线程结束
currentThreadCount--; //处理线程队列减少一个
}
}
}
/**
* 当线程处理完成后重新放到空闲线程队列中.
* @param c ControlRunnable
*/
protected synchronized void returnController(ControlRunnable c) {
if (0 == currentThreadCount || stopThePool) { //如果线程池关闭或当前连接线程数量为0
c.terminate(); //关闭当前线程.
return;
}
currentThreadsBusy--; //处理线程队列的数量减少一个
pool.addElement(c); //空闲线程队列中增加一个
notifyAll(); //唤醒可能在等待连接的线程.
}
/**
* 当一个处理线程出现异常时,要重新开启一个空闭线程.,并唤醒在等待空闲线程的线程.ThreadPool的runIt中等待的线程.
*/
protected synchronized void notifyThreadEnd() {
currentThreadsBusy--; //因从线程是在处理数据时出现异常,所处理线程队列的数量要减一个.
currentThreadCount--; //因出现异常的线程关闭了.所开户线程的数量要减少一个.
notifyAll(); //唤醒等待连接的阻塞线程.
openThreads(minSpareThreads); //重新打开minSpareThreads个线程.如currentThreadCount的数量大于minSpareThreads,则还是不开启新线程.
}
/**
* 调整各种线程队列数量
*/
protected void adjustLimits() {
if (maxThreads <= 0) { //如果最大线程数小于0
maxThreads = MAX_THREADS; //设置最大线程数为100
}
if (maxSpareThreads >= maxThreads) { //如果最大多余线程数大于最大线程数.
maxSpareThreads = maxThreads; //设置最大多余线程数为最大线程数.
}
if (maxSpareThreads <= 0) { //如果最大多余线程数小于0
if (1 == maxThreads) {
maxSpareThreads = 1; //如最大线程数为1的情况下,设置最大多余线程数为1.
}
else {
maxSpareThreads = maxThreads / 2; //设置最大多余线程数为最大线程数的一半.
}
}
if (minSpareThreads > maxSpareThreads) { //如果最小多余线程大于最大多余线程数
minSpareThreads = maxSpareThreads; //设置最小多余线程数为最大多余线程数.
}
if (minSpareThreads <= 0) { //如果最小多余线程数小于0
if (1 == maxSpareThreads) {
minSpareThreads = 1; //如最大线程数为1的情况下,则设置最小多余线程数为1.
}
else {
minSpareThreads = maxSpareThreads / 2; //否则设置最小多余线程数为最大多余线程数的一半.
}
}
}
/**
* 打开指定数量的空闲线程队列
* @param toOpen int
*/
protected void openThreads(int toOpen) { //toOpen=minSpareThreads
if (toOpen > maxThreads) {
toOpen = maxThreads;
}
if (0 == currentThreadCount) { //如果当前线程池中的线程数量为0
pool = new Vector(toOpen); //创建一个有minSpareThreads数量的Vector
}
//因第二次增加时对第一次增加的线程不能重复增加.所要从currentThreadCount开始.
for (int i = currentThreadCount; i < toOpen; i++) { //先增加minSparethreads数量的线程.
pool.addElement(new ControlRunnable(this)); //Runnable实例对象,可用于创建线程
}
currentThreadCount = toOpen;
}
/**
* 监视线程,用于监听当前空闲线程是否大于最大多余线程数量,如存在则关闭多余的空闲线程.
*/
class MonitorRunnable
implements Runnable {
ThreadPool p;
Thread t;
boolean shouldTerminate;
/**
* construct
* @param p ThreadPool
*/
MonitorRunnable(ThreadPool p) {
shouldTerminate = false;
this.p = p;
t = new Thread(this);
t.start();
}
public void run() {
while (true) {
try {
synchronized (this) {
this.wait(WORK_WAIT_TIMEOUT);
}
if (shouldTerminate) { //如果结束
break;
}
p.checkSpareControllers(); //检查是否有多余线程.
}
catch (Throwable t) {
t.printStackTrace();
}
}
}
public void stop() {
this.terminate();
}
public synchronized void terminate() {
shouldTerminate = true;
this.notifyAll();
}
}
} |
|
发表于 2011-7-16 13:45:35
QQ好友和群
腾讯微博
腾讯朋友
微信
转播
淘帖
收藏
支持
反对