`
sfjsffjjj
  • 浏览: 11189 次
  • 性别: Icon_minigender_1
  • 来自: 北京
社区版块
存档分类
最新评论

线程池示例代码，请大家多指教

多线程工作Springthread设计模式 
阅读更多
包结构
src
  test
    TestThreadPool          测试类
  thread
    ThreadPool              线程池类
    WorkThread              工作线程类
    TaskMonitorThread       任务监测线程类
    TaskTimeOutThread       任务超时监测线程类
  task
    TaskManager             任务管理器
    WorkTask                任务接口
    WorkTaskImp             正常任务类
    WorkTaskAImp            异常任务类
    WorkTaskBImp            超时任务类
  event
    AbstractEvent           任务事件类
    BeginTaskEvent          任务执行开始事件类
    EndTaskEvent            任务执行结束事件类
    TaskRunTime             任务运行时间类
    TaskTimeOutEvent        任务执行超时事件类

源代码 
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

  • src.rar (5.4 KB)
  • 下载次数: 390
分享到：
| oracle几种关闭方式
评论
56 楼 belover 2009-09-05  
贴大段的代码。实在是懒得看。建议：把UML 时序图，活动图，类图。
设计清晰一目了然。细节的实现。就是 技术人员 本身素质的问题了。

还有一点。的确如楼上所说。Think In Java 第四版
55 楼 xingzhaozhong 2009-08-31  
yangke_love 写道
JDK5.0中有相关的线程管理内容，你可以参考下Think In Java 第四版

是这样的，Thinking In Java写的很详细了
54 楼 snowhui 2009-08-31  
如果你用是ssh在添加spring ,hibernate ,struts你就可以建连接池。
53 楼 vtrtbb 2009-08-31  
zhang_jie439 写道
import java.util.*;

public class ThreadPool {
	private LinkedList<Runnable> taskList = null;
	private boolean closed = false;

	public ThreadPool() {
		this.taskList = new LinkedList<Runnable>();
		for (int i = 0; i < 4; i++) {
			Thread work = new WorkThread();
			work.setDaemon(true);
			work.start();
		}
	}

	public synchronized void addTask(Runnable task) {
		if (closed)
			return;
		if (null == task)
			return;
		this.taskList.add(task);
		notify();
	}

	private synchronized Runnable getTask() {
		while (this.taskList.size() == 0) {
			try {
				if (closed)	return null;
				wait();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		return this.taskList.removeFirst();
	}

	public void close() {
		synchronized (this) {
			this.closed = true;
		}

	}

	private class WorkThread extends Thread {
		public void run() {
			while (true) {
				Runnable task = null;
				task = getTask();
				if (null == task)
					return;
				task.run();
			}
		}
	}
}


哥们，这个多少年前就存在吧！！！！
52 楼 anranran 2009-08-26  
别整那么多SLEEP,wait下,合适的机会notify.SLeep不好.
51 楼 C_J 2009-08-25  
楼上用的啥数据库，为什么自己做个ConnectionPool？？
50 楼 qiushyfm 2009-08-25   
package com.shy.sql;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.Enumeration;
import java.util.Hashtable;

public class QConnectionPool {
	private Hashtable connections;// 连接列表
	private int increment;// 连接增长量
	private String dbURL;// 数据库连接url
	private String user;
	private String password;
	private String diverClassName;// 驱动名称
	/**
	 * 构造方法
	 * 
	 * @param dbURL
	 *            数据库连接url
	 * @param user
	 *            用户名
	 * @param password
	 *            密码
	 * @param driverClassName
	 *            驱动名称
	 * @param initialConnections
	 *            初始连接数
	 * @param increment
	 *            增加数
	 * @throws SQLException
	 * @throws ClassNotFoundException
	 */
	public QConnectionPool(String dbURL, String user, String password,
			String driverClassName, int initialConnections, int increment)
			throws SQLException, ClassNotFoundException {
		Class.forName(driverClassName);
		this.dbURL = dbURL;
		this.user = user;
		this.password = password;
		this.increment = increment;
		connections = new Hashtable();
		for (int i = 0; i < initialConnections; i++) {
			connections.put(DriverManager.getConnection(dbURL, user, password),
					false);
		}
	}

	/**
	 * 从连接池返回一个Connection
	 * @return
	 * @throws SQLException
	 */
	public Connection getConnection() throws SQLException {
		Connection conn = null;
		Enumeration conns = connections.keys();
		synchronized (connections) {
			while (conns.hasMoreElements()) {
				conn = (Connection) conns.nextElement();
				Boolean b = (Boolean) connections.get(conn);
				//如果是不使用的连接，使其使用
				if (b == Boolean.FALSE) {
					try {
						conn.setAutoCommit(true);
					} catch (SQLException e) {
						connections.remove(conn);
						conn = DriverManager.getConnection(dbURL, user,
								password);
					}
					// 更新Hashtable
					connections.put(conn, Boolean.TRUE);

					return conn;
				}
			}
		}

		//如果没有空闲的连接，增加连接数
		for (int i = 0; i < increment; i++) {
			connections.put(DriverManager.getConnection(dbURL, user, password),
					Boolean.TRUE);
		}
		return getConnection();
	}
	
	/**
	 * 由连接池客户用来向连接池返回一个Connection
	 */
	public void returnConnection(Connection returned){
		if(connections.containsKey(returned)) {
			connections.put(returned, Boolean.FALSE);
		}
	}
}

共享一个自己的，呵呵！
49 楼 sfjsffjjj 2009-08-25  
引用
楼主的线程池还是处于一个比较初级的阶段，并且其中有一些问题的

凤舞凰扬兄，能说的详细点吗
48 楼 凤舞凰扬 2009-08-25  
   一般来说，在处理大量请求，希望通过多线程并发处理的方式提高效率的时候。比如最常见的线程池就是HTTP线程池。
47 楼 snowhui 2009-08-25  
我想知道在什么情况下用线程池呢。本人用java的手法写过一个.net连接池，性能还可以！
但线程池，我实在没有 什么概念，能告诉我在什么时用吗》
46 楼 凤舞凰扬 2009-08-24  
    呵呵，楼主的线程池还是处于一个比较初级的阶段，并且其中有一些问题的。
    如果楼主是用JDK1.5以上的版本，建议你使用BlockingQueue吧。
45 楼 vinter 2009-08-24  
zhang_jie439 会导致死锁吧
44 楼 fjlyxx 2009-08-22  
楼上的GG 模式能够解决你的问题  线程池给你 如果你只看到 获取 和 返回线程的接口 那么你就不能从根本上解决竞争的问题  你的这个状态对象 也就没办法得到有效的控制.
43 楼 C_J 2009-08-21  
xujunJ2EE 写道
上限阀值和下限阀值很难设定的，因为并发请求往往是很不稳定的。
当你使用ReferenceObjectPool时，GC何时回收对象你无法确定，有得必有失，这是无法避免的
spring用单例是因为大部分放置在springIOC容器中的都是粗粒度组件对象，这些往往都是无状态的。
common pool创建的对象怎么会是无状态对象呢？我看你对状态还没有理解。比如我创建一个student对象，其中有一个name属性，A线程拿到后setName("xujun"),name的状态就变化了。而且，可以肯定的说，放入pool的只会是有状态对象。无状态的就应该用单例



是因为我看到了commons pool的一个pool的borrow()实现如下： 

    public synchronized Object borrowObject() throws Exception {
        assertOpen();
        Object obj = null;
        boolean newlyCreated = false;
    
        
        while (null == obj) {
            if (!_pool.empty()) {
                obj = _pool.pop();// _pool is a stack!!!
...


每次取对象的时候都是从栈顶取，不可能取到以前的对象的呀？？返回的都是经过清理的“新”对象呀


所以对于 pool来说，怎么存在“状态”对象呢？





42 楼 xujunJ2EE 2009-08-21  
最后提醒你一下：你也提到了“脏对象”。如果是无状态对象，那么怎么会存在“脏对象”呢。嘿嘿
41 楼 xujunJ2EE 2009-08-21  
上限阀值和下限阀值很难设定的，因为并发请求往往是很不稳定的。
当你使用ReferenceObjectPool时，GC何时回收对象你无法确定，有得必有失，这是无法避免的
spring用单例是因为大部分放置在springIOC容器中的都是粗粒度组件对象，这些往往都是无状态的。
common pool创建的对象怎么会是无状态对象呢？我看你对状态还没有理解。比如我创建一个student对象，其中有一个name属性，A线程拿到后setName("xujun"),name的状态就变化了。而且，可以肯定的说，放入pool的只会是有状态对象。无状态的就应该用单例
40 楼 xujunJ2EE 2009-08-21  
对象有状态还是否使用pool没有关系吧，比如Connection对象就是一个有状态对象
39 楼 C_J 2009-08-20  
xujunJ2EE 写道
我个人认为没有必要，象LZ那样的是专门的Thread对象的线程池，所以他可以自己定义一个workThread对象，但如果你是开发一个类似common pool这样的通用对象池，你根本不知道放入你池中的对象是什么，你怎么能让放入你pool中的对象都实现一个state的属性呢，一种方式就是要求放入你pool的对象都实现一个interface中的读写state方法，这样就侵入原始对象了。还有一种就是在pool中放入原始对象的proxy对象，在proxy中定义state。而且你还要考虑对象的activate(), passivate()等处理，否则你从pool中拿到的就是一个脏对象，一般来说我们放对象池的都是有状态对象，无状态对象完全可以用单例来作，没有必要pool化。
common pool很多的思想还是很优秀的，他通过一个factory来剥离这些操作，让你的原始对象完全无侵入，对于需要放入pool的对象，你实现一个factory，在factory中处理状态操作和对象产生，这样将这些可能导致侵入的方法都剥离了。
至于common pool没有实现wati(),我想这并不是一个bug，而且在某些情况下，这是一个很好的考虑，thread的等待往往可能要化更多的时间，这个时间可能比创建一个对象更多。所以，common pool的设计就是给你一个pool，pool里面都是可复用的对象，当你不够用的时候就创建新对象，够用的时候就用pool里面的对象。在rod的without EJB中也说过，对象pool有时候不是好事，当并发请求的线程小于pool的size，那么请求线程在堵塞，当pool比size大的时候，pool实际上有点浪费。所以spring默认都是单例。我觉得common pool在设计的时候可能想到了这个因素，当并发线程大于pool的时候，干脆就创建对象。



是的,你提到的大致4个问题:

1,commons pool中的StatckObjectPool 和 ReferenceObjectPool等 本身设计的确实不允许"脏"对象的,像你说的,外围的activate(), passivate()充当了"清理"工作..

2,
引用
一般来说我们放对象池的都是有状态对象，无状态对象完全可以用单例来作，没有必要pool化。


真的是这样么? 既然对象有状态,还有必要用pool么? 我觉得这有点违背pool想法吧?且会混乱业务逻辑和抽象逻辑的,这点也是我现在比较迷惑的地方...

3,如果pool的_maxSleeping阀值设置不当的话，在多线程APP下确实会导致堵塞...上限阀值和下限阀值很重要吧...


4,另外commons提供了一个ReferenceObjectPool,这种的好处在于内存不足时,GC无条件回收pool中的空闲对象,.这样资源可适当得到利用..不过pool中总要有一定量的闲对象，不可能每次都new吧。

5,至于spring用单例的原因,我想是因为spring本身就是一个lightweight组件,Rob应该也很不喜欢weight..


另外 我看到的commons中,是在borrow()时，如果pool中没有空闲obj的时候,就创建新的obj,当然这个新对象是无状态的，所以建立在pool存储也需要建立在"无状态对象"的基本之上...
38 楼 xujunJ2EE 2009-08-20  
如果你的对象确实很重量级，确实需要限制对象数量，那么可以稍微修改一下common pool的代码，通过activeNum来判断活动对象数量，以下代码是我修改common pool的源码，已测试通过；
Object obj = null;
        //如果Pool中有对象就直接给付
        if(pool.size()>0)
        {
            obj = pool.remove(0);
            System.out.println("线程直接获得对象");
        }
        /*
         *如果Pool中没有对象，要区分两种情况：
         *1 没有到达最大对象数，继续创建一个新的对象
         *2 到达对象最大数了，等待其他线程释放对象
         */
        else
        {
            //没有到达最大对象数，继续创建一个新的对象
            if(activeNum < parameter.getMaxNum())
            {
                obj = objFactory.makeObject();
                System.out.println("线程new获得对象");
            }
            else
            {
                try {
                    System.out.println("线程在等待!");
                    this.wait();
                } catch (InterruptedException ex) {
                    throw new Exception("线程异常");
                }
                obj = pool.firstElement();
                System.out.println("线程等待结束，获得其他线程return的对象");
            }
        }
        //对象的状态设置，并在任何发生异常的情况下钝化对象状态，取消引用
        try
        {
            //设置该对象为活动状态
            objFactory.activateObject(obj);
            //对象检验
            if(!objFactory.validateObject(obj))
            {
                throw new Exception("ValidateObject failed");
            }
            activeNum ++;
        }
        catch(Throwable t)
        {
            objFactory.destroyObject(obj);
            obj = null;
            return obj;
        }
       
        return obj;
    }

    public synchronized  void returnObject(Object obj)
    {
        //先钝化对象，清除状态
        boolean passivate = false;
        try
        {
            if(objFactory.validateObject(obj))
            {
                objFactory.passivateObject(obj);
                passivate = true;
            }
        }
        catch(Exception ex)
        {
            passivate = false;
        }
        //如果钝化成功，将对象放入到对象池中
        if(passivate)
        {
            //如果对象池满了，直接销毁该对象
            if(pool.size() >= parameter.getMaxNum())
            {
                objFactory.destroyObject(obj);
            }
            else
            {
                pool.add(obj);
            }
            //通知wait
            notifyAll();
            activeNum--;
        }
37 楼 xujunJ2EE 2009-08-20  
我个人认为没有必要，象LZ那样的是专门的Thread对象的线程池，所以他可以自己定义一个workThread对象，但如果你是开发一个类似common pool这样的通用对象池，你根本不知道放入你池中的对象是什么，你怎么能让放入你pool中的对象都实现一个state的属性呢，一种方式就是要求放入你pool的对象都实现一个interface中的读写state方法，这样就侵入原始对象了。还有一种就是在pool中放入原始对象的proxy对象，在proxy中定义state。而且你还要考虑对象的activate(), passivate()等处理，否则你从pool中拿到的就是一个脏对象，一般来说我们放对象池的都是有状态对象，无状态对象完全可以用单例来作，没有必要pool化。
common pool很多的思想还是很优秀的，他通过一个factory来剥离这些操作，让你的原始对象完全无侵入，对于需要放入pool的对象，你实现一个factory，在factory中处理状态操作和对象产生，这样将这些可能导致侵入的方法都剥离了。
至于common pool没有实现wati(),我想这并不是一个bug，而且在某些情况下，这是一个很好的考虑，thread的等待往往可能要化更多的时间，这个时间可能比创建一个对象更多。所以，common pool的设计就是给你一个pool，pool里面都是可复用的对象，当你不够用的时候就创建新对象，够用的时候就用pool里面的对象。在rod的without EJB中也说过，对象pool有时候不是好事，当并发请求的线程小于pool的size，那么请求线程在堵塞，当pool比size大的时候，pool实际上有点浪费。所以spring默认都是单例。我觉得common pool在设计的时候可能想到了这个因素，当并发线程大于pool的时候，干脆就创建对象。
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

Magicbox
Global site tag (gtag.js) - Google Analytics