一个虽复杂但可直接套用的线程池实例
周sir |
2015年4月10日 |
多线程编程 |
0 条评论 | 1515
继「多线程程序模型研究」之后，再给出一个更复杂、但功能比较完整、可以直接套用的线程池。实例改写自 Apache 项目源码（原文约 800 行，跑在 Tomcat 等服务器上），这里做了精简，只留核心，并配测试类。
学完可以直接模仿套用，不必自己从零再写一遍池。关键两块：线程池类 ThreadPool 和 接口 ThreadPoolRunnable。
一、ThreadPool：可伸缩的工人池
文件 ThreadPool.java 两百多行。先看完整源码，再拆 openThreads、runIt 以及两个内部类。
import java.util.Vector;
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;
protected int maxThreads;
protected int minSpareThreads;
protected int maxSpareThreads;
protected int currentThreadCount;
protected int currentThreadsBusy;
protected boolean stopThePool;
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);
}
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;
}
public void runIt(ThreadPoolRunnable r) {
if (null == r) {
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;
openThreads(toOpen);
} 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++;
}
c.runIt(r);
}
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++) {
ControlRunnable c = (ControlRunnable) pool.firstElement();
pool.removeElement(c);
c.terminate();
currentThreadCount--;
}
}
}
protected synchronized void returnController(ControlRunnable c) {
if (0 == currentThreadCount || stopThePool) {
c.terminate();
return;
}
currentThreadsBusy--;
pool.addElement(c);
notify();
}
protected synchronized void notifyThreadEnd() {
currentThreadsBusy--;
currentThreadCount--;
notify();
openThreads(minSpareThreads);
}
protected void adjustLimits() {
if (maxThreads <= 0) {
maxThreads = MAX_THREADS;
}
if (maxSpareThreads >= maxThreads) {
maxSpareThreads = maxThreads;
}
if (maxSpareThreads <= 0) {
if (1 == maxThreads) {
maxSpareThreads = 1;
} else {
maxSpareThreads = maxThreads / 2;
}
}
if (minSpareThreads > maxSpareThreads) {
minSpareThreads = maxSpareThreads;
}
if (minSpareThreads <= 0) {
if (1 == maxSpareThreads) {
minSpareThreads = 1;
} else {
minSpareThreads = maxSpareThreads / 2;
}
}
}
protected void openThreads(int toOpen) {
if (toOpen > maxThreads) {
toOpen = maxThreads;
}
if (0 == currentThreadCount) {
pool = new Vector(toOpen);
}
for (int i = currentThreadCount; i < toOpen; i++) {
pool.addElement(new ControlRunnable(this));
}
currentThreadCount = toOpen;
}
class MonitorRunnable implements Runnable {
ThreadPool p;
Thread t;
boolean shouldTerminate;
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 synchronized void terminate() {
shouldTerminate = true;
this.notify();
}
}
class ControlRunnable implements Runnable {
ThreadPool p;
Thread t;
ThreadPoolRunnable toRun;
boolean shouldTerminate;
boolean shouldRun;
boolean noThData;
Object thData[] = null;
ControlRunnable(ThreadPool p) {
toRun = null;
shouldTerminate = false;
shouldRun = false;
this.p = p;
t = new Thread(this);
t.start();
noThData = true;
thData = null;
}
public void run() {
while (true) {
try {
synchronized (this) {
if (!shouldRun && !shouldTerminate) {
this.wait();
}
}
if (shouldTerminate) {
break;
}
try {
if (noThData) {
thData = toRun.getInitData();
noThData = false;
}
if (shouldRun) {
toRun.runIt(thData);
}
} catch (Throwable t) {
System.err.println("ControlRunnable Throwable: ");
t.printStackTrace();
shouldTerminate = true;
shouldRun = false;
p.notifyThreadEnd();
} finally {
if (shouldRun) {
shouldRun = false;
p.returnController(this);
}
}
if (shouldTerminate) {
break;
}
} catch (InterruptedException ie) {
}
}
}
public synchronized void runIt(ThreadPoolRunnable toRun) {
if (toRun == null) {
throw new NullPointerException("No Runnable");
}
this.toRun = toRun;
shouldRun = true;
this.notify();
}
public synchronized void terminate() {
shouldTerminate = true;
this.notify();
}
}
}
启动时往 Vector 里装 minSpareThreads 个 ControlRunnable，构造方法里就会 t.start()。若 shouldRun 和 shouldTerminate 都是 false，工人就 wait；shouldRun 为 true 时调用任务的 runIt(Object[])——这就是你要覆盖的方法。
minSpareThreads 个工人都 Busy 之后，池会再创建 minSpareThreads 个。MonitorRunnable 每隔 60 秒（WORK_WAIT_TIMEOUT）调一次 checkSpareControllers：若空闲工人多于 maxSpareThreads，就 terminate 多出来的空闲线程。
| 参数 |
默认 |
含义 |
| MAX_THREADS / maxThreads |
100 |
池上限 |
| MAX_SPARE_THREADS / maxSpareThreads |
50 |
空闲工人上限，多了就收回 |
| MIN_SPARE_THREADS / minSpareThreads |
10 |
启动时预创建，忙时也按这个粒度扩 |
| WORK_WAIT_TIMEOUT |
60s |
监视线程检查空闲的间隔 |
二、任务接口 ThreadPoolRunnable
接口只有两个空方法：getInitData 和 runIt。自己写一个任务类实现它，把活放进 runIt。不用初始化数据就让 getInitData 返回 null。
public interface ThreadPoolRunnable {
public Object[] getInitData();
public void runIt(Object thData[]);
}
三、套用：还是扫端口
测试类 TestThreadPool.java，任务仍是扫描端口（参见「多线程程序模型研究」）：
import java.net.InetAddress;
import java.net.Socket;
public class TestThreadPool {
public static void main(String[] args) {
String host = null; //第一个参数，目标主机。
int beginport = 1; //第二个参数，开始端口。
int endport = 65535; //第三个参数，结束端口。
try{
host = args[0];
beginport = Integer.parseInt(args[1]);
endport = Integer.parseInt(args[2]);
if(beginport <= 0 || endport >= 65536 || beginport > endport){
throw new Exception("Port is illegal");
}
}catch(Exception e){
System.out.println("Usage: java PortScannerSingleThread host beginport endport");
System.exit(0);
}
ThreadPool tp = new ThreadPool();
tp.setMaxThreads(100);
tp.setMaxSpareThreads(50);
tp.setMinSpareThreads(10);
tp.start();
for(int i = beginport; i <= endport; i++){
Task task = new Task(host,i);
tp.runIt(task);
}
}
}
class Task implements ThreadPoolRunnable{
String host;
int port;
Task(String host, int port){
this.host = host;
this.port = port;
}
public Object[] getInitData(){
return null;
}
public void runIt(Object thData[]){
Socket s = null;
try{
s = new Socket(InetAddress.getByName(host),port);
System.out.println("The port " + port + " at " + host + " is open.");
}catch(Exception e){
}finally{
try{
if(s != null) s.close();
}catch(Exception e){
}
}
}
}
main 里三个参数：目标主机、起止端口。然后 new ThreadPool()，用 setter 调 maxThreads / maxSpareThreads / minSpareThreads，再 start()。Task 实现接口，在 runIt 里 new Socket 完成扫描。
JDK 1.4.2 编译运行通过。输入 java TestThreadPool 10.1.1.182 1 10000：
The port 25 at 10.1.1.182 is open.
The port 110 at 10.1.1.182 is open.
The port 135 at 10.1.1.182 is open.
...
把 ThreadPool 和 ThreadPoolRunnable 直接拷进自己的项目即可：任务类实现接口，runIt 里写业务。
一句话总结：工人不够就按 minSpare 扩，空闲太多就按 maxSpare 收；你的任务只需实现 runIt。
转载请注明来源：一个虽复杂但可直接套用的线程池实例
我是周sir，这是我的博客。致力于分享我学会的技术和优秀的文章。微信公众号 “周sir专栏” 欢迎大家关注。
