Java实现通用线程池
2007-09-29 15:16
381 查看
URL: http://blog.csdn.net/polarman/archive/2006/08/09/1042149.aspx
线程池通俗的描述就是预先创建若干空闲线程，等到需要用多线程去处理事务的时候去唤醒某些空闲线程执行处理任务，这样就省去了频繁创建线程的时间，因为频 繁创建线程是要耗费大量的CPU资源的。如果一个应用程序需要频繁地处理大量并发事务，不断的创建销毁线程往往会大大地降低系统的效率，这时候线程池就派 上用场了。
本文旨在使用Java语言编写一个通用的线程池。当需要使用线程池处理事务时，只需按照指定规范封装好事务处理对象，然后用已有的线程池对象去自动选择空 闲线程自动调用事务处理对象即可。并实现线程池的动态修改（修改当前线程数，最大线程数等）。下面是实现代码：
//ThreadTask .java
package polarman.threadpool;
/** *//**
* 线程任务
* @author ryang
* 2006-8-8
*/
public interface ThreadTask ...{
public void run();
}
//PooledThread.java
package polarman.threadpool;
import java.util.Collection;
import java.util.Vector;
/** *//**
* 接受线程池管理的线程
* @author ryang
* 2006-8-8
*/
public class PooledThread extends Thread ...{
protected Vector tasks = new Vector();
protected boolean running = false;
protected boolean stopped = false;
protected boolean paused = false;
protected boolean killed = false;
private ThreadPool pool;
public PooledThread(ThreadPool pool)...{
this.pool = pool;
}
public void putTask(ThreadTask task)...{
tasks.add(task);
}
public void putTasks(ThreadTask[] tasks)...{
for(int i=0; i<tasks.length; i++)
this.tasks.add(tasks[i]);
}
public void putTasks(Collection tasks)...{
this.tasks.addAll(tasks);
}
protected ThreadTask popTask()...{
if(tasks.size() > 0)
return (ThreadTask)tasks.remove(0);
else
return null;
}
public boolean isRunning()...{
return running;
}
public void stopTasks()...{
stopped = true;
}
public void stopTasksSync()...{
stopTasks();
while(isRunning())...{
try ...{
sleep(5);
} catch (InterruptedException e) ...{
}
}
}
public void pauseTasks()...{
paused = true;
}
public void pauseTasksSync()...{
pauseTasks();
while(isRunning())...{
try ...{
sleep(5);
} catch (InterruptedException e) ...{
}
}
}
public void kill()...{
if(!running)
interrupt();
else
killed = true;
}
public void killSync()...{
kill();
while(isAlive())...{
try ...{
sleep(5);
} catch (InterruptedException e) ...{
}
}
}
public synchronized void startTasks()...{
running = true;
this.notify();
}
public synchronized void run()...{
try...{
while(true)...{
if(!running || tasks.size() == 0)...{
pool.notifyForIdleThread();
//System.out.println(Thread.currentThread().getId() + ": 空闲");
this.wait();
}else...{
ThreadTask task;
while((task = popTask()) != null)...{
task.run();
if(stopped)...{
stopped = false;
if(tasks.size() > 0)...{
tasks.clear();
System.out.println(Thread.currentThread().getId() + ": Tasks are stopped");
break;
}
}
if(paused)...{
paused = false;
if(tasks.size() > 0)...{
System.out.println(Thread.currentThread().getId() + ": Tasks are paused");
break;
}
}
}
running = false;
}
if(killed)...{
killed = false;
break;
}
}
}catch(InterruptedException e)...{
return;
}
//System.out.println(Thread.currentThread().getId() + ": Killed");
}
}
//ThreadPool.java
package polarman.threadpool;
import java.util.Collection;
import java.util.Iterator;
import java.util.Vector;
/** *//**
* 线程池
* @author ryang
* 2006-8-8
*/
public class ThreadPool ...{
protected int maxPoolSize;
protected int initPoolSize;
protected Vector threads = new Vector();
protected boolean initialized = false;
protected boolean hasIdleThread = false;
public ThreadPool(int maxPoolSize, int initPoolSize)...{
this.maxPoolSize = maxPoolSize;
this.initPoolSize = initPoolSize;
}
public void init()...{
initialized = true;
for(int i=0; i<initPoolSize; i++)...{
PooledThread thread = new PooledThread(this);
thread.start();
threads.add(thread);
}
//System.out.println("线程池初始化结束，线程数=" + threads.size() + " 最大线程数=" + maxPoolSize);
}
public void setMaxPoolSize(int maxPoolSize)...{
//System.out.println("重设最大线程数，最大线程数=" + maxPoolSize);
this.maxPoolSize = maxPoolSize;
if(maxPoolSize < getPoolSize())
setPoolSize(maxPoolSize);
}
/** *//**
* 重设当前线程数
* 若需杀掉某线程，线程不会立刻杀掉，而会等到线程中的事务处理完成
* 但此方法会立刻从线程池中移除该线程，不会等待事务处理结束
* @param size
*/
public void setPoolSize(int size)...{
if(!initialized)...{
initPoolSize = size;
return;
}else if(size > getPoolSize())...{
for(int i=getPoolSize(); i<size && i<maxPoolSize; i++)...{
PooledThread thread = new PooledThread(this);
thread.start();
threads.add(thread);
}
}else if(size < getPoolSize())...{
while(getPoolSize() > size)...{
PooledThread th = (PooledThread)threads.remove(0);
th.kill();
}
}
//System.out.println("重设线程数，线程数=" + threads.size());
}
public int getPoolSize()...{
return threads.size();
}
protected void notifyForIdleThread()...{
hasIdleThread = true;
}
protected boolean waitForIdleThread()...{
hasIdleThread = false;
while(!hasIdleThread && getPoolSize() >= maxPoolSize)...{
try ...{
Thread.sleep(5);
} catch (InterruptedException e) ...{
return false;
}
}
return true;
}
public synchronized PooledThread getIdleThread()...{
while(true)...{
for(Iterator itr=threads.iterator(); itr.hasNext();)...{
PooledThread th = (PooledThread)itr.next();
if(!th.isRunning())
return th;
}
if(getPoolSize() < maxPoolSize)...{
PooledThread thread = new PooledThread(this);
thread.start();
threads.add(thread);
return thread;
}
//System.out.println("线程池已满，等待...");
if(waitForIdleThread() == false)
return null;
}
}
public void processTask(ThreadTask task)...{
PooledThread th = getIdleThread();
if(th != null)...{
th.putTask(task);
th.startTasks();
}
}
public void processTasksInSingleThread(ThreadTask[] tasks)...{
PooledThread th = getIdleThread();
if(th != null)...{
th.putTasks(tasks);
th.startTasks();
}
}
public void processTasksInSingleThread(Collection tasks)...{
PooledThread th = getIdleThread();
if(th != null)...{
th.putTasks(tasks);
th.startTasks();
}
}
}
下面是线程池的测试程序
//ThreadPoolTest.java
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import polarman.threadpool.ThreadPool;
import polarman.threadpool.ThreadTask;
public class ThreadPoolTest ...{
public static void main(String[] args) ...{
System.out.println(""quit" 退出");
System.out.println(""task A 10" 启动任务A，时长为10秒");
System.out.println(""size 2" 设置当前线程池大小为2");
System.out.println(""max 3" 设置线程池最大线程数为3");
System.out.println();
final ThreadPool pool = new ThreadPool(3, 2);
pool.init();
Thread cmdThread = new Thread()...{
public void run()...{
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
while(true)...{
try ...{
String line = reader.readLine();
String words[] = line.split(" ");
if(words[0].equalsIgnoreCase("quit"))...{
System.exit(0);
}else if(words[0].equalsIgnoreCase("size") && words.length >= 2)...{
try...{
int size = Integer.parseInt(words[1]);
pool.setPoolSize(size);
}catch(Exception e)...{
}
}else if(words[0].equalsIgnoreCase("max") && words.length >= 2)...{
try...{
int max = Integer.parseInt(words[1]);
pool.setMaxPoolSize(max);
}catch(Exception e)...{
}
}else if(words[0].equalsIgnoreCase("task") && words.length >= 3)...{
try...{
int timelen = Integer.parseInt(words[2]);
SimpleTask task = new SimpleTask(words[1], timelen * 1000);
pool.processTask(task);
}catch(Exception e)...{
}
}
} catch (IOException e) ...{
e.printStackTrace();
}
}
}
};
cmdThread.start();
/**//*
for(int i=0; i<10; i++){
SimpleTask task = new SimpleTask("Task" + i, (i+10)*1000);
pool.processTask(task);
}*/
}
}
class SimpleTask implements ThreadTask...{
private String taskName;
private int timeLen;
public SimpleTask(String taskName, int timeLen)...{
this.taskName = taskName;
this.timeLen = timeLen;
}
public void run() ...{
System.out.println(Thread.currentThread().getId() +
": START TASK "" + taskName + """);
try ...{
Thread.sleep(timeLen);
} catch (InterruptedException e) ...{
}
System.out.println(Thread.currentThread().getId() +
": END TASK "" + taskName + """);
}
}
使用此线程池相当简单，下面两行代码初始化线程池：
ThreadPool pool = new ThreadPool(3, 2);
pool.init();
要处理的任务实现ThreadTask...接口即可（如测试代码里的SimpleTask），这个接口只有一个方法run()
两行代码即可调用：
ThreadTask task = ... //实例化你的任务对象
pool.processTask(task);
线程池通俗的描述就是预先创建若干空闲线程，等到需要用多线程去处理事务的时候去唤醒某些空闲线程执行处理任务，这样就省去了频繁创建线程的时间，因为频 繁创建线程是要耗费大量的CPU资源的。如果一个应用程序需要频繁地处理大量并发事务，不断的创建销毁线程往往会大大地降低系统的效率，这时候线程池就派 上用场了。
本文旨在使用Java语言编写一个通用的线程池。当需要使用线程池处理事务时，只需按照指定规范封装好事务处理对象，然后用已有的线程池对象去自动选择空 闲线程自动调用事务处理对象即可。并实现线程池的动态修改（修改当前线程数，最大线程数等）。下面是实现代码：
//ThreadTask .java
package polarman.threadpool;
/** *//**
* 线程任务
* @author ryang
* 2006-8-8
*/
public interface ThreadTask ...{
public void run();
}
//PooledThread.java
package polarman.threadpool;
import java.util.Collection;
import java.util.Vector;
/** *//**
* 接受线程池管理的线程
* @author ryang
* 2006-8-8
*/
public class PooledThread extends Thread ...{
protected Vector tasks = new Vector();
protected boolean running = false;
protected boolean stopped = false;
protected boolean paused = false;
protected boolean killed = false;
private ThreadPool pool;
public PooledThread(ThreadPool pool)...{
this.pool = pool;
}
public void putTask(ThreadTask task)...{
tasks.add(task);
}
public void putTasks(ThreadTask[] tasks)...{
for(int i=0; i<tasks.length; i++)
this.tasks.add(tasks[i]);
}
public void putTasks(Collection tasks)...{
this.tasks.addAll(tasks);
}
protected ThreadTask popTask()...{
if(tasks.size() > 0)
return (ThreadTask)tasks.remove(0);
else
return null;
}
public boolean isRunning()...{
return running;
}
public void stopTasks()...{
stopped = true;
}
public void stopTasksSync()...{
stopTasks();
while(isRunning())...{
try ...{
sleep(5);
} catch (InterruptedException e) ...{
}
}
}
public void pauseTasks()...{
paused = true;
}
public void pauseTasksSync()...{
pauseTasks();
while(isRunning())...{
try ...{
sleep(5);
} catch (InterruptedException e) ...{
}
}
}
public void kill()...{
if(!running)
interrupt();
else
killed = true;
}
public void killSync()...{
kill();
while(isAlive())...{
try ...{
sleep(5);
} catch (InterruptedException e) ...{
}
}
}
public synchronized void startTasks()...{
running = true;
this.notify();
}
public synchronized void run()...{
try...{
while(true)...{
if(!running || tasks.size() == 0)...{
pool.notifyForIdleThread();
//System.out.println(Thread.currentThread().getId() + ": 空闲");
this.wait();
}else...{
ThreadTask task;
while((task = popTask()) != null)...{
task.run();
if(stopped)...{
stopped = false;
if(tasks.size() > 0)...{
tasks.clear();
System.out.println(Thread.currentThread().getId() + ": Tasks are stopped");
break;
}
}
if(paused)...{
paused = false;
if(tasks.size() > 0)...{
System.out.println(Thread.currentThread().getId() + ": Tasks are paused");
break;
}
}
}
running = false;
}
if(killed)...{
killed = false;
break;
}
}
}catch(InterruptedException e)...{
return;
}
//System.out.println(Thread.currentThread().getId() + ": Killed");
}
}
//ThreadPool.java
package polarman.threadpool;
import java.util.Collection;
import java.util.Iterator;
import java.util.Vector;
/** *//**
* 线程池
* @author ryang
* 2006-8-8
*/
public class ThreadPool ...{
protected int maxPoolSize;
protected int initPoolSize;
protected Vector threads = new Vector();
protected boolean initialized = false;
protected boolean hasIdleThread = false;
public ThreadPool(int maxPoolSize, int initPoolSize)...{
this.maxPoolSize = maxPoolSize;
this.initPoolSize = initPoolSize;
}
public void init()...{
initialized = true;
for(int i=0; i<initPoolSize; i++)...{
PooledThread thread = new PooledThread(this);
thread.start();
threads.add(thread);
}
//System.out.println("线程池初始化结束，线程数=" + threads.size() + " 最大线程数=" + maxPoolSize);
}
public void setMaxPoolSize(int maxPoolSize)...{
//System.out.println("重设最大线程数，最大线程数=" + maxPoolSize);
this.maxPoolSize = maxPoolSize;
if(maxPoolSize < getPoolSize())
setPoolSize(maxPoolSize);
}
/** *//**
* 重设当前线程数
* 若需杀掉某线程，线程不会立刻杀掉，而会等到线程中的事务处理完成
* 但此方法会立刻从线程池中移除该线程，不会等待事务处理结束
* @param size
*/
public void setPoolSize(int size)...{
if(!initialized)...{
initPoolSize = size;
return;
}else if(size > getPoolSize())...{
for(int i=getPoolSize(); i<size && i<maxPoolSize; i++)...{
PooledThread thread = new PooledThread(this);
thread.start();
threads.add(thread);
}
}else if(size < getPoolSize())...{
while(getPoolSize() > size)...{
PooledThread th = (PooledThread)threads.remove(0);
th.kill();
}
}
//System.out.println("重设线程数，线程数=" + threads.size());
}
public int getPoolSize()...{
return threads.size();
}
protected void notifyForIdleThread()...{
hasIdleThread = true;
}
protected boolean waitForIdleThread()...{
hasIdleThread = false;
while(!hasIdleThread && getPoolSize() >= maxPoolSize)...{
try ...{
Thread.sleep(5);
} catch (InterruptedException e) ...{
return false;
}
}
return true;
}
public synchronized PooledThread getIdleThread()...{
while(true)...{
for(Iterator itr=threads.iterator(); itr.hasNext();)...{
PooledThread th = (PooledThread)itr.next();
if(!th.isRunning())
return th;
}
if(getPoolSize() < maxPoolSize)...{
PooledThread thread = new PooledThread(this);
thread.start();
threads.add(thread);
return thread;
}
//System.out.println("线程池已满，等待...");
if(waitForIdleThread() == false)
return null;
}
}
public void processTask(ThreadTask task)...{
PooledThread th = getIdleThread();
if(th != null)...{
th.putTask(task);
th.startTasks();
}
}
public void processTasksInSingleThread(ThreadTask[] tasks)...{
PooledThread th = getIdleThread();
if(th != null)...{
th.putTasks(tasks);
th.startTasks();
}
}
public void processTasksInSingleThread(Collection tasks)...{
PooledThread th = getIdleThread();
if(th != null)...{
th.putTasks(tasks);
th.startTasks();
}
}
}
下面是线程池的测试程序
//ThreadPoolTest.java
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import polarman.threadpool.ThreadPool;
import polarman.threadpool.ThreadTask;
public class ThreadPoolTest ...{
public static void main(String[] args) ...{
System.out.println(""quit" 退出");
System.out.println(""task A 10" 启动任务A，时长为10秒");
System.out.println(""size 2" 设置当前线程池大小为2");
System.out.println(""max 3" 设置线程池最大线程数为3");
System.out.println();
final ThreadPool pool = new ThreadPool(3, 2);
pool.init();
Thread cmdThread = new Thread()...{
public void run()...{
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
while(true)...{
try ...{
String line = reader.readLine();
String words[] = line.split(" ");
if(words[0].equalsIgnoreCase("quit"))...{
System.exit(0);
}else if(words[0].equalsIgnoreCase("size") && words.length >= 2)...{
try...{
int size = Integer.parseInt(words[1]);
pool.setPoolSize(size);
}catch(Exception e)...{
}
}else if(words[0].equalsIgnoreCase("max") && words.length >= 2)...{
try...{
int max = Integer.parseInt(words[1]);
pool.setMaxPoolSize(max);
}catch(Exception e)...{
}
}else if(words[0].equalsIgnoreCase("task") && words.length >= 3)...{
try...{
int timelen = Integer.parseInt(words[2]);
SimpleTask task = new SimpleTask(words[1], timelen * 1000);
pool.processTask(task);
}catch(Exception e)...{
}
}
} catch (IOException e) ...{
e.printStackTrace();
}
}
}
};
cmdThread.start();
/**//*
for(int i=0; i<10; i++){
SimpleTask task = new SimpleTask("Task" + i, (i+10)*1000);
pool.processTask(task);
}*/
}
}
class SimpleTask implements ThreadTask...{
private String taskName;
private int timeLen;
public SimpleTask(String taskName, int timeLen)...{
this.taskName = taskName;
this.timeLen = timeLen;
}
public void run() ...{
System.out.println(Thread.currentThread().getId() +
": START TASK "" + taskName + """);
try ...{
Thread.sleep(timeLen);
} catch (InterruptedException e) ...{
}
System.out.println(Thread.currentThread().getId() +
": END TASK "" + taskName + """);
}
}
使用此线程池相当简单，下面两行代码初始化线程池：
ThreadPool pool = new ThreadPool(3, 2);
pool.init();
要处理的任务实现ThreadTask...接口即可（如测试代码里的SimpleTask），这个接口只有一个方法run()
两行代码即可调用：
ThreadTask task = ... //实例化你的任务对象
pool.processTask(task);
内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理 
相关文章推荐
- java利用线程池实现处理socket请求的小例子
- 通用线程池的设计和实现[C语言]
- Java 并发编程深入学习——线程池及其实现原理
- Java 线程池的原理与实现
- Java 线程池的原理与实现
- Java 多线程编程之九：使用 Executors 和 ThreadPoolExecutor 实现的 Java 线程池的例子
- JSP实现的Java通用简单分页算法
- 运用JAVA的concurrent.ExecutorService线程池实现socket的TCP和UDP连接
- 【java】HttpClient实现HTTP文件通用下载工具类
- 使用 java 开源资源实现通用可靠的下载组件
- Java手写线程池实现
- Java线程池有助于实现最佳资源利用率
- java 实现视频转换通用工具类：获取视频元数据信息（一）
- Java的Executor框架和线程池实现原理
- Android java 实现自己线程池
- Java 多线程：分析线程池的实现原理
- java之线程池简单实现
- java 实现视频转换通用工具类：视频加水印-Ffmpeg（五-1）
- JAVA线程池的简单实现及优先级设置[转]
- 通用 Java 文件上传和下载组件的设计与实现