CaptainCook
- 浏览: 137018 次
- 性别:
- 来自: 北京
-
最新评论
-
Oo注定oO幸福：
怎么我不能运行的？是不是还有东东没有设置的？
Frameset布局 -
hjceo：
一看就是百度文库下的
SSH实现上传下载 -
yahon：
不错，回顾一下
Frameset布局 -
zhou_wu：
样例还不错，简单学习了
Frameset布局 -
mengzhiang：
看完了，感觉你太厉害了，我是一个普通大学毕业的小本，大学四年全 ...
张龙慢慢程序人生路
包结构
src
test
TestThreadPool 测试类
thread
ThreadPool 线程池类
WorkThread 工作线程类
TaskMonitorThread 任务监测线程类
TaskTimeOutThread 任务超时监测线程类
task
TaskManager 任务管理器
WorkTask 任务接口
WorkTaskImp 正常任务类
WorkTaskAImp 异常任务类
WorkTaskBImp 超时任务类
event
AbstractEvent 任务事件类
BeginTaskEvent 任务执行开始事件类
EndTaskEvent 任务执行结束事件类
TaskRunTime 任务运行时间类
TaskTimeOutEvent 任务执行超时事件类
源代码
Java代码
package test;
import task.TaskManager;
import task.WorkTask;
import task.WorkTaskAImp;
import task.WorkTaskBImp;
import task.WorkTaskImp;
import thread.ThreadPool;
/**
* 线程池测试类,测试功能如下：
* 1、测试线程池创建功能
* 2、测试处理并发请求功能
* 3、测试关闭功能
**/
public class TestThreadPool {
public static void main(String[] args){
//创建线程池,开启处理请求服务
final int threadCount=10;
ThreadPool pool=ThreadPool.getInstance();
pool.init(threadCount);
//接收客户端请求
WorkTask task1=new WorkTaskBImp("执行超时任务...");
TaskManager.addTask(task1);
final int requestCount=20;
for(int i=0;i<requestCount;i++){
WorkTask task=new WorkTaskImp("执行第"+i+"个增加用户操作...");
TaskManager.addTask(task);
}
WorkTask task2=new WorkTaskBImp("执行超时任务...");
TaskManager.addTask(task2);
for(int i=0;i<requestCount;i++){
WorkTask task=new WorkTaskAImp("执行第"+i+"个修改用户异常操作...");
TaskManager.addTask(task);
}
for(int i=0;i<requestCount;i++){
WorkTask task=new WorkTaskImp("执行第"+i+"个删除用户操作...");
TaskManager.addTask(task);
}
//关闭线程池
try {
Thread.sleep(2000);//为了显示处理请求效果
pool.close();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
package thread;
import java.util.ArrayList;
import java.util.List;
import java.util.Vector;
import event.BeginTaskEvent;
import event.EndTaskEvent;
import task.TaskManager;
/**
* 线程池类，功能如下：
* 1、初始化线程池
* 2、获取空闲线程
* 3、任务运行，注册超时监测
* 4、任务结束，注销超时监测
* 5、关闭线程池
*/
public class ThreadPool {
private int threadcount;
private int GetIdleThreadPollTime=50;//获取空闲线程轮询间隔时间,可配置
private static ThreadPool pool=new ThreadPool();//线程实例
private Vector<WorkThread> threadlist=new Vector<WorkThread>();//工作线程列表
private TaskMonitorThread mainThread;//任务监测线程
private TaskTimeOutThread timeThread; //任务超时线程
private boolean StopGetIdleThread=false;
//单例模式
private ThreadPool(){
}
public static synchronized ThreadPool getInstance(){
return pool;
}
private void stopGetIdleThread(){
StopGetIdleThread = true;
}
//初始化线程池
public void init(int count){
System.out.println("开始初始化线程池...");
threadcount=count;
for(int i=0;i<count;i++){
WorkThread t=new WorkThread(new Integer(i));
threadlist.add(t);
t.start();
}
mainThread=new TaskMonitorThread(pool);
mainThread.start();
timeThread=new TaskTimeOutThread(pool);
timeThread.start();
System.out.println("结束初始化线程池...");
}
//获取空闲线程
public WorkThread getIdleThread(){
while(true){
if (StopGetIdleThread) return null;
synchronized(threadlist){
for(int i=0;i<threadlist.size();i++){
WorkThread t=(WorkThread)threadlist.get(i);
if (t.getMyState().equals(WorkThread.IDlESTATE)){
return t;
}
}
}
try {
Thread.sleep(GetIdleThreadPollTime);//放弃CPU,若干时间后重新获取空闲线程
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
//任务运行，注册监测
public void beginTaskRun(BeginTaskEvent begin){
timeThread.beginTaskRun(begin);
}
//任务结束，注销监视
public void endTaskRun(EndTaskEvent end){
timeThread.endTaskRun(end);
}
//从工作线程表中移除线程
public void removeWorkThread(WorkThread t){
threadlist.remove(t);
}
//添加新的线程
public void addWorkThread(){
synchronized(threadlist){
WorkThread t=new WorkThread(new Integer(++threadcount));
threadlist.add(t);
t.start();
}
}
//关闭线程池
public void close(){
//停止获取空闲线程
stopGetIdleThread();
//关闭任务监测线程，不再接收请求
mainThread.kill();
//关闭超时监测线程
timeThread.kill();
//关闭工作线程，不再处理任务
for(int i=0;i<threadlist.size();i++){
WorkThread t=(WorkThread)threadlist.get(i);
t.kill();
}
}
}
package thread;
import task.WorkTask;
import event.BeginTaskEvent;
import event.EndTaskEvent;
/**
* 工作线程类,功能如下：
* 1、执行业务方法，业务参数可动态设置
* 2、自身状态可设置、可获取
* 3、自我唤醒功能
* 4、自杀功能
*/
public final class WorkThread extends Thread{
private boolean shutdown=false;
private String info; //业务参数
private Object threadKey;//线程标识
private Object lock=new Object();//锁对象
private String state; //线程状态
private int waitExecFinishPollTime=500;//关闭线程时的轮询等待时间，可配置
public static final String CREATESTATE="1";//创建状态
public static final String RUNSTATE="2"; //运行状态
public static final String IDlESTATE="3"; //空闲状态
private WorkTask nowTask; //当前任务
//获取线程标识key
public Object getThreadKey() {
return threadKey;
}
//设置线程的任务
public void setWorkTask(WorkTask task){
this.nowTask=task;
}
//设置是否关闭线程
private void setShutdown(boolean shutdown) {
this.shutdown = shutdown;
}
//设置线程状态
private void setMyState(String state){
this.state=state;
}
//获取线程状态
public String getMyState(){
return state;
}
public WorkThread(Object key){
System.out.println("正在创建工作线程...线程编号"+key.toString());
this.threadKey=key;
this.state=CREATESTATE;
}
@Override
public synchronized void start() {
// TODO Auto-generated method stub
super.start();
setMyState(RUNSTATE);
}
public void run(){
while(true){
try {
setMyState(IDlESTATE);
synchronized(this){
wait(); /*开始等待，直至被激活*/
}
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
if (shutdown) break;
try{
new BeginTaskEvent(this,Thread.currentThread(),nowTask).execute();
nowTask.execute();//执行业务
new EndTaskEvent(this,Thread.currentThread(),nowTask).execute();
}catch(Exception e){
new EndTaskEvent(this,Thread.currentThread(),nowTask).execute();
System.out.println(e.getMessage());
}
}
}
//重新激活线程
public void activate(){
synchronized(this){
setMyState(RUNSTATE);
notify();
}
}
//关闭线程
public void kill(){
synchronized(this){
if (this.getMyState().equals(IDlESTATE)){//如果线程处于空闲状态,则直接关掉
System.out.println("正在关闭工作线程...线程编号"+threadKey.toString());
this.setShutdown(true);
this.activate();
}else if (this.getMyState().equals(RUNSTATE)){//如果线程处于运行状态,则执行完后再关掉
System.out.println("正在等待线程执行业务完成...工作线程编号"+threadKey.toString());
while(this.getMyState().equals(RUNSTATE)){
try {
Thread.sleep(waitExecFinishPollTime);//放弃CPU,若干时间后再检查线程状态
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
System.out.println("正在关闭工作线程...线程编号"+threadKey.toString());
this.setShutdown(true);
this.activate();
}
}
}
}
package thread;
import task.TaskManager;
import task.WorkTask;
/**
* 任务检测线程类
* 1、自杀功能
*/
public final class TaskMonitorThread extends Thread {
private ThreadPool threadPool;
private int GetWorkTaskPollTime=10;//监测任务轮询时间，可配置
private boolean shutdown=false;
public TaskMonitorThread(ThreadPool pool){
System.out.println("正在创建任务监测线程...");
this.threadPool=pool;
}
private void setShutDown(boolean b){
this.shutdown=b;
}
@Override
public void run() {
// TODO Auto-generated method stub
while(true){
if (shutdown) break;
WorkTask task=TaskManager.getWorkTask();//看是否有任务请求
if (task==null){
try {
Thread.sleep(GetWorkTaskPollTime);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}else{
WorkThread t=threadPool.getIdleThread();//获取空闲线程
if (t==null) break;
t.setWorkTask(task);//设置线程任务
task.setTaskThreadKey(t.getThreadKey());//为了显示任务当前线程
t.activate();//激活空闲线程
try {
Thread.sleep(GetWorkTaskPollTime);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
//关闭线程
public void kill(){
System.out.println("正在关闭任务监测线程...");
this.setShutDown(true);
}
}
package thread;
import java.util.Vector;
import event.BeginTaskEvent;
import event.EndTaskEvent;
import event.TaskRunTime;
import event.TaskTimeOutEvent;
/**
* 任务超时监测线程类
* 1、任务开始注册
* 2、任务完成注销
* 3、自杀功能
*/
public class TaskTimeOutThread extends Thread {
private ThreadPool pool;
private boolean shutdown=false;
private Vector<TaskRunTime> taskruntimelist=new Vector<TaskRunTime>();//运行任务列表
private int pollTime=500; //轮询时间
private int TaskOutTime=2000; //任务过时时间
public TaskTimeOutThread(ThreadPool pool){
this.pool=pool;
}
@Override
public void run() {
// TODO Auto-generated method stub
while(!shutdown){
synchronized(taskruntimelist){
for(int i=0;i<taskruntimelist.size();i++){
TaskRunTime t=(TaskRunTime) taskruntimelist.get(i);
if (t.checkRunTimeOut(TaskOutTime)){
taskruntimelist.remove(i);
new TaskTimeOutEvent(t.getEvent()).execute();
break;
}
}
}
try {
sleep(pollTime);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
//任务运行，开始监测
public void beginTaskRun(BeginTaskEvent begin){
taskruntimelist.add(new TaskRunTime(begin));
}
//任务正常结束
public void endTaskRun(EndTaskEvent end){
synchronized(taskruntimelist){
for(int i=0;i<taskruntimelist.size();i++){
BeginTaskEvent begin=((TaskRunTime) taskruntimelist.get(i)).getEvent();
if (begin.equals(end)){
taskruntimelist.remove(i);
break;
}
}
}
}
//自杀
public void kill(){
System.out.println("正在关闭超时监测线程...");
while(taskruntimelist.size()>0){
try {
Thread.sleep(pollTime);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
shutdown=true;
}
}
package task;
import java.util.ArrayList;
import java.util.List;
/**
* 任务管理器
* 1、添加任务
* 2、监测是否有新任务
*/
public class TaskManager {
private static List taskQueue=new ArrayList<WorkTask>(); //任务队列
private TaskManager(){
}
//添加任务
public synchronized static void addTask(WorkTask task){
taskQueue.add(task);
}
//判断是否有任务未执行
public synchronized static WorkTask getWorkTask(){
if (taskQueue.size()>0){
return (WorkTask)taskQueue.remove(0);
}else
return null;
}
}
package task;
/**
* 任务接口
* 继承它来定义自己具体的工作任务
*/
public interface WorkTask {
void execute() throws Exception; //执行工作任务
void setTaskThreadKey(Object key);//设置任务线程编号
}
package task;
/**
* 任务类1
* 正常执行的工作任务
*/
public class WorkTaskImp implements WorkTask {
protected String param;
protected Object threadkey; //为了显示执行线程编号
protected final int TaskExecTime=500; //任务执行时间
public void execute() throws Exception {
// TODO Auto-generated method stub
System.out.println(param+"工作线程编号"+threadkey.toString());
Thread.sleep(TaskExecTime);
}
public WorkTaskImp(String param){
this.param=param;
}
public void setTaskThreadKey(Object key){
this.threadkey=key;
}
public String toString(){
return param+"工作线程编号"+threadkey.toString();
}
}
package task;
/**
* 任务类2
* 执行报异常的工作任务
*/
public class WorkTaskAImp extends WorkTaskImp{
public WorkTaskAImp(String param) {
super(param);
// TODO Auto-generated constructor stub
}
public void execute() throws Exception {
// TODO Auto-generated method stub
throw new Exception("运行WorkTaskAImp任务时出错");
}
}
package task;
/*
* 任务类3
* 执行超时的工作任务
*/
public class WorkTaskBImp extends WorkTaskImp{
public WorkTaskBImp(String param) {
super(param);
// TODO Auto-generated constructor stub
}
public void execute() throws Exception {
// TODO Auto-generated method stub
System.out.println("正在"+param);
Thread.sleep(50000); //随便定义
}
}
package event;
import task.WorkTask;
import thread.WorkThread;
/*
*任务抽象事件
*/
public abstract class AbstractEvent {
protected WorkThread workthread;
protected Thread nowthread;
protected WorkTask nowtask;
//事件触发
public synchronized void execute(){};
@Override
public boolean equals(Object obj) {
// TODO Auto-generated method stub
AbstractEvent other=(AbstractEvent)obj;
return this.workthread==other.workthread&&this.nowtask==this.nowtask;
};
}
package event;
import task.WorkTask;
import thread.ThreadPool;
import thread.WorkThread;
/*
* 任务开始运行事件
*/
public class BeginTaskEvent extends AbstractEvent{
public BeginTaskEvent(WorkThread workthread,Thread nowthread,WorkTask task){
this.workthread=workthread;
this.nowthread=nowthread;
this.nowtask=task;
}
@Override
public void execute() {
// TODO Auto-generated method stub
ThreadPool pool=ThreadPool.getInstance();
pool.beginTaskRun(this);
}
}
package event;
import task.WorkTask;
import thread.ThreadPool;
import thread.WorkThread;
/*
* 任务运行结束事件
*/
public class EndTaskEvent extends AbstractEvent {
public EndTaskEvent(WorkThread workthread,Thread nowthread,WorkTask task){
this.workthread=workthread;
this.nowthread=nowthread;
this.nowtask=task;
}
@Override
public void execute() {
// TODO Auto-generated method stub
ThreadPool pool=ThreadPool.getInstance();
pool.endTaskRun(this);
}
}
package event;
/*
* 任务运行时间类
*/
public class TaskRunTime {
private long begintime;
private long endtime;
private BeginTaskEvent event;
public TaskRunTime(BeginTaskEvent event){
this.event=event;
this.begintime=System.currentTimeMillis();
this.endtime=this.begintime;
}
public BeginTaskEvent getEvent() {
return event;
}
//检查是否超时
public boolean checkRunTimeOut(long maxtime){
endtime=System.currentTimeMillis();
long cha=endtime-begintime;
return cha>=maxtime;
}
}
package event;
import task.WorkTask;
import thread.ThreadPool;
import thread.WorkThread;
/*
* 任务超时事件
*/
public class TaskTimeOutEvent {
private AbstractEvent event;
public TaskTimeOutEvent(AbstractEvent event){
this.event=event;
}
@SuppressWarnings("deprecation")
public void execute() {
// TODO Auto-generated method stub
ThreadPool pool=ThreadPool.getInstance();
pool.addWorkThread();
pool.removeWorkThread(event.workthread);
Object obj=event.workthread.getThreadKey();
System.out.println("正在停止工作超时线程...线程编号"+obj);
event.nowthread.stop();
}
}
运行结果
开始初始化线程池...
正在创建工作线程...线程编号0
正在创建工作线程...线程编号1
正在创建工作线程...线程编号2
正在创建工作线程...线程编号3
正在创建工作线程...线程编号4
正在创建工作线程...线程编号5
正在创建工作线程...线程编号6
正在创建工作线程...线程编号7
正在创建工作线程...线程编号8
正在创建工作线程...线程编号9
正在创建任务监测线程...
结束初始化线程池...
正在执行超时任务1...
执行第0个增加用户操作...工作线程编号1
执行第1个增加用户操作...工作线程编号2
执行第2个增加用户操作...工作线程编号3
执行第3个增加用户操作...工作线程编号4
执行第4个增加用户操作...工作线程编号5
执行第5个增加用户操作...工作线程编号6
执行第6个增加用户操作...工作线程编号7
执行第7个增加用户操作...工作线程编号8
执行第8个增加用户操作...工作线程编号9
执行第9个增加用户操作...工作线程编号1
执行第10个增加用户操作...工作线程编号2
执行第11个增加用户操作...工作线程编号3
执行第12个增加用户操作...工作线程编号4
执行第13个增加用户操作...工作线程编号5
执行第14个增加用户操作...工作线程编号6
执行第15个增加用户操作...工作线程编号7
执行第16个增加用户操作...工作线程编号8
执行第17个增加用户操作...工作线程编号9
执行第18个增加用户操作...工作线程编号1
执行第19个增加用户操作...工作线程编号2
正在执行超时任务2...
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
运行WorkTaskAImp任务时出错
执行第0个删除用户操作...工作线程编号4
执行第1个删除用户操作...工作线程编号5
执行第2个删除用户操作...工作线程编号6
执行第3个删除用户操作...工作线程编号7
执行第4个删除用户操作...工作线程编号8
执行第5个删除用户操作...工作线程编号9
执行第6个删除用户操作...工作线程编号1
执行第7个删除用户操作...工作线程编号2
执行第8个删除用户操作...工作线程编号4
执行第9个删除用户操作...工作线程编号5
执行第10个删除用户操作...工作线程编号6
执行第11个删除用户操作...工作线程编号7
执行第12个删除用户操作...工作线程编号8
执行第13个删除用户操作...工作线程编号9
正在关闭任务监测线程...
正在关闭超时监测线程...
正在创建工作线程...线程编号11
正在停止工作超时线程...线程编号0
正在创建工作线程...线程编号12
正在停止工作超时线程...线程编号3
正在关闭工作线程...线程编号1
正在关闭工作线程...线程编号2
正在关闭工作线程...线程编号4
正在关闭工作线程...线程编号5
正在关闭工作线程...线程编号6
正在关闭工作线程...线程编号7
正在关闭工作线程...线程编号8
正在关闭工作线程...线程编号9
正在关闭工作线程...线程编号11
正在关闭工作线程...线程编号12
分享到：
发表评论
-
运行jar应用程序引用其他jar包的四种方法
2009-10-18 11:47 2044大家都知道一个java应用项目可以打包成一个jar，当然你 ... -
图解classloader加载class的流程及自定义ClassLoader
2009-10-18 11:45 1430/** * 转载请注明作者longdick ht ... -
回调函数的理解
2009-10-12 00:48 948面使用java回调函数来实现一个测试函数运行时间的工具类： ... -
深入探索Java工作原理：JVM内存回收及其他
2009-06-05 12:49 1342Java语言引入了Java虚拟 ... -
Hibernate缓存机制
2009-05-02 12:35 932缓存是位于应用程序与 ... -
Java 10年回顾:世界各地开发高手谈Java
2009-04-29 17:00 790到今年五月，Java已经走过了它辉煌的十年。我们邀请了世界各地 ... -
多线程
2009-04-17 17:46 854public class TestThread { p ... -
javamail发送邮件
2009-04-17 15:37 1245今天学习了一下JavaMail，javamail发送邮件确实是 ...
相关推荐
NULL 博文链接：https://yulincqupt.iteye.com/blog/1673919
线程池的作用: 线程池作用就是限制系统中执行线程的数量。 根据系统的环境情况，可以自动或手动设置线程数量，达到运行的最佳效果；少了浪费了系统资源，多了造成系统拥挤效率不高。用线程池控制线程数量，其他...
比较完整的线程池实例代码，Linux系统下完美运行
包含DEMO（ThreadPoolDemo.rar），源代码 thread_pool.rar(只有一个HPP文件）及经过裁减的Boost1.32库；之后以用Boost是因为ThreadPool中用到了 Boost库的线程组件。 关键字：thread,pool,线程池
一个简单的线程池示例，可以自定义线程数量和执行任务，代码简洁可扩展性强。在使用上也很方便。下面是一个简单的调用 int main() { xcyk::ThreadPool threadPool("xcyk"); SYSTEM_INFO SystemInfo; ...
这是基于c#的线程池示例，应用于多线程等等
简单演示了如何使用线程池，可以再源代码的基础上进行修改进行更高级的操作，适合用于学习
取自卢伟的专栏DELPHI 线程池代码(http://blog.csdn.net/babyvspp/archive/2008/01/01/2008234.aspx)，封装的十分完美，用法也非常简单，可以根据使用者设定的初始工作线程数和最大线程数以及线程执行时间自动增加...
Java 线程池示例 一个简单的 Java 线程池示例
g++ -o test quickstart.cpp -lboost_system -lboost_thread -lpthread
C实现的线程池,示例展示如何在应用程序中使用线程池，以及实现线程池可以如何提高应用程序的性能。
封装好的线程池稳定版本，可直接使用，包含线程池的库代码和一个简单的使用示例。对于一般任务，性能上足够了。
Linux高级程序设计 第三版 源码（示例代码）（杨宗德）,和课本上的代码是完全一样的，希望大家下载。
内容索引:VC/C++源码,系统相关,连接池,多线程 VC++写的线程池示例，运行后可以看到线程的数目在变化，截图如上，VC编程朋友参阅一下。
主要介绍了Spring Boot整合FTPClient线程池的实现示例，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
通过1个简单的VC例程，演示了线程池的使用方法，以及线程池的函数用法。代码为VS2015工程，可直接打开编译。
TCP SOCKET多线程通信，使用.NET自带线程池，高效率 ，可用来测试端口防火墙等，注释详细，
读书笔记和示例代码 包括 线程基础/同步sync/并发concurrent/容器/线程池 示例代码 JUC容器的源码解析 Synchronized 同步容器 Concurrent 并发容器: Map/List/Queue 阻塞队列 JUC锁: CountDownLatch 闭锁 AQS 锁的...