tomcat线程池的实现

bobbai

　　Tomcat的线程池主要使用了ThreadPool.java、ThreadPoolRunnable.java、ThreadWithAttributes.java，其中ThreadPoolRunnable.java是一个接口，所有的需要使用线程池的服务都可以实现这个接口。而实现的核心则再ThreadPool.java中的两个内部类：MonitorRunnable.java和ControlRunnable.java。
　　MonitorRunnable.java在线程池启动之后定期（60s）的扫描线程数，如果空闲的线程大于最大空闲线程数，则结束多余的线程。
　　ControlRunnable.java是所有启动的线程，若由任务需要执行，ThreadPool会先找一个空闲的ControlRunnable来执行，若没有空闲的，则创建，若现有的busy线程已经达到最大值，则等待。任务执行结束后通知ControlRunnable继续wait,直到有任务执行或被MonitorRunnable回收。
　　若要使用线程池可以实现Runnable接口，或者可以实现ThreadPoolRunnable 接口，当然自己还可以扩展这个类，以便实现更多的使用线程池的方式。
　　ThreadPool.java
　　


package com.xiao.tomcatthreadpool;
import java.util.Hashtable;
public class ThreadPool {
public static final int MAX_THREADS = 10;
public static final int MAX_THREADS_MIN = 4;
public static final int MAX_SPARE_THREADS = 5;
public static final int MIN_SPARE_THREADS = 2;
public static final int WORK_WAIT_TIMEOUT = 10*1000;
private String name = "TP";
private boolean isDaemon;
private boolean stopThePool;
private int maxSpareThreads;
private int minSpareThreads;
private int currentThreadCount;
private int currentThreadsBusy;
private int maxThreads;
private int threadPriority = Thread.NORM_PRIORITY;
private int sequence = 0;
private ControlRunnable[] pool;
private MonitorRunnable monitor;
protected Hashtable threads=new Hashtable();
public ThreadPool() {
maxThreads = MAX_THREADS;
maxSpareThreads = MAX_SPARE_THREADS;
minSpareThreads = MIN_SPARE_THREADS;
currentThreadCount = 0;
currentThreadsBusy = 0;
stopThePool = false;
}
public static ThreadPool createThreadPool() {
return new ThreadPool();
}
public synchronized void start() {
stopThePool = false;
currentThreadCount = 0;
currentThreadsBusy = 0;
pool = new ControlRunnable[maxThreads];
openThreads(minSpareThreads);
if (maxSpareThreads < maxThreads) {
monitor = new MonitorRunnable(this);
}
}
public void run(Runnable r) {
ControlRunnable c = findControlRunnable();
c.runIt(r);
}
public void runIt(ThreadPoolRunnable r) {
if(null == r) {
throw new NullPointerException();
}
ControlRunnable c = findControlRunnable();
c.runIt(r);
}
public ControlRunnable findControlRunnable() {
ControlRunnable c;
if ( stopThePool ) {
throw new IllegalStateException();
}
synchronized(this) {
while (currentThreadsBusy == currentThreadCount) {
if (currentThreadCount < maxThreads) {
int toOpen = currentThreadCount + minSpareThreads;
openThreads(toOpen);
} else {
try {
this.wait();
}
catch(InterruptedException e) {
}
}
}
if(0 == currentThreadCount || stopThePool) {
throw new IllegalStateException();
}
int pos = currentThreadCount - currentThreadsBusy - 1;
c = pool[pos];
pool[pos] = null;
currentThreadsBusy++;
}
return c;
}
public void openThreads(int toOpen) {
if(toOpen > maxThreads) {
toOpen = maxThreads;
}
for(int i=currentThreadCount; i<toOpen; i++) {
pool[i - currentThreadsBusy] = new ControlRunnable(this);
}
currentThreadCount = toOpen;
}
public void addThread(ThreadWithAttributes t, ControlRunnable r) {
threads.put(t, r);
}
public void removeThread(Thread t) {
threads.remove(t);
}
public synchronized void notifyThreadEnd(ControlRunnable r) {
currentThreadCount --;
currentThreadsBusy --;
notify();
}
public synchronized void returnController(ControlRunnable r) {
if(0 == currentThreadCount || stopThePool) {
r.terminate();
return;
}
currentThreadsBusy--;
pool[currentThreadCount - currentThreadsBusy - 1] = r;
notify();
}
public synchronized void checkSpareControllers() {
if(stopThePool) {
return;
}
if((currentThreadCount - currentThreadsBusy) > maxSpareThreads) {
int toFree = currentThreadCount - currentThreadsBusy - maxSpareThreads;
for(int i=0; i<toFree; i++) {
ControlRunnable cr = pool[currentThreadCount-currentThreadsBusy -1];
cr.terminate();
pool[currentThreadCount-currentThreadsBusy -1] = null;
currentThreadCount --;
}
}
}
/**
* MonitorRunnable主要任务是监控线程数
* 如果线程数大于最大线程则回收线程
*/
public static class MonitorRunnable implements Runnable {
ThreadPool tp;
Thread t;
boolean shouldTerminate;
int interval = WORK_WAIT_TIMEOUT;
public MonitorRunnable(ThreadPool tp) {
this.tp = tp;
this.start();
}
public void setInterval(int i) {
interval = i;
}
public void start() {
shouldTerminate = false;
t = new Thread(this);
t.setDaemon(tp.getDaemon());
t.setName(tp.getName() + "-Monitor");
t.start();
}
public void stop() {
terminal();
}
public synchronized void terminal() {
this.shouldTerminate = true;
this.notify();
}
public void run() {
while(true) {
try {
synchronized(this) {
this.wait(interval);
}
if(shouldTerminate) {
break;
}
//System.out.println("currentThreadCount=" + currentThreadCount + " currentThreadsBusy=" + currentThreadsBusy + " ");
tp.checkSpareControllers();
} catch(InterruptedException e) {
}
}
}
}
public static class ControlRunnable implements Runnable {
private ThreadPool tp;
private boolean shouldTerminate;
private ThreadWithAttributes     t;
private ThreadPoolRunnable   toRun;
private Runnable toRunRunnable;
private boolean    shouldRun;
public ControlRunnable(ThreadPool tp) {
toRun = null;
shouldTerminate = false;
shouldRun = false;
this.tp = tp;
t = new ThreadWithAttributes(tp, this);
t.setDaemon(true);
t.setName(tp.getName() + "-Processor" + tp.incSequence());
t.setPriority(tp.getThreadPriority());
tp.addThread(t, this);
t.start();
}
public void run() {
boolean _shouldRun = false;
boolean _shouldTerminate = false;
ThreadPoolRunnable _toRun = null;
try {
while(true) {
try {
synchronized(this) {
System.out.println("shouldRun=" + shouldRun);
while(!shouldRun && !shouldTerminate) {
this.wait();
}
_shouldRun = shouldRun;
_shouldTerminate = shouldTerminate;
_toRun = toRun;
}
if (_shouldTerminate)
break;
try {
if(_shouldRun) {
if (_toRun != null) {
_toRun.runIt(t.getThreadData(tp));
} else if (toRunRunnable != null) {
toRunRunnable.run();
} else {
}
}
} catch(Throwable r) {
_shouldTerminate = true;
_shouldRun = false;
tp.notifyThreadEnd(this);
} finally {
if(_shouldRun) {
shouldRun = false;
tp.returnController(this);
}
}
if (_shouldTerminate) {
break;
}
} catch(InterruptedException e) {
}
}
} finally {
tp.removeThread(Thread.currentThread());
}
}
public synchronized void runIt(Runnable toRun) {
this.toRunRunnable = toRun;
shouldRun = true;
this.notify();
}
public synchronized void runIt(ThreadPoolRunnable toRun) {
this.toRun = toRun;
shouldRun = true;
this.notify();
}
public void stop() {
this.terminate();
}
public void kill() {
t.stop();
}
public synchronized void terminate() {
shouldTerminate = true;
this.notify();
}
}
public String getName() {
return name;
}
public boolean getDaemon() {
return isDaemon;
}
public int getThreadPriority() {
return threadPriority;
}
public int incSequence() {
return sequence ++;
}
public void setThreadPriority(int threadPriority) {
this.threadPriority = threadPriority;
}
}
　　ThreadWithAttributes.java
　　


package com.xiao.tomcatthreadpool;
import java.util.Hashtable;
public class ThreadWithAttributes extends Thread {
private Object control;
public static int MAX_NOTES=16;
private Object notes[]=new Object[MAX_NOTES];
private Hashtable attributes=new Hashtable();
private String currentStage;
private Object param;
private Object thData[];
public ThreadWithAttributes(Object control, Runnable r) {
super(r);
this.control=control;
}
public final Object[] getThreadData(Object control ) {
return thData;
}
public final void setThreadData(Object control, Object thData[] ) {
this.thData=thData;
}
public final void setNote( Object control, int id, Object value ) {
if( this.control != control ) return;
notes[id]=value;
}
public final String getCurrentStage(Object control) {
if( this.control != control ) return null;
return currentStage;
}
/** Information about the current request ( or the main object
* we are processing )
*/
public final Object getParam(Object control) {
if( this.control != control ) return null;
return param;
}
public final void setCurrentStage(Object control, String currentStage) {
if( this.control != control ) return;
this.currentStage = currentStage;
}
public final void setParam( Object control, Object param ) {
if( this.control != control ) return;
this.param=param;
}
public final Object getNote(Object control, int id ) {
if( this.control != control ) return null;
return notes[id];
}
public final Hashtable getAttributes(Object control) {
if( this.control != control ) return null;
return attributes;
}
}

　　ThreadPoolRunnable.java
　　


package com.xiao.tomcatthreadpool;
public interface ThreadPoolRunnable {
public Object[] getInitData();
public void runIt(Object thData[]);
}