比较完整数据源实现

 

 

package com.bpsoft.blank.jdbc;

import java.io.InputStream;
import java.io.PrintWriter;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.Enumeration;
import java.util.Properties;
import java.util.Vector;

import org.apache.log4j.Logger;

import com.bpsoft.blank.util.LogUtil;



public class KDataSource implements javax.sql.DataSource{
	/**驱动*/
	private String driver;
	/**url*/
	private String url;
	/**用户名*/
	private String user;
	/**密码*/
	private String password;
	/**初始化连接数*/
	private int initCount=10;
	/**连接池中最大连接数*/
	private int maxCount=50;
	/**当前使用连接数*/
	private int currentCount=0;
	/**等待连接最长时间*/
	private int wait=3000;
	/** 保持有效连接数 */
	private int maxActive=5;
	/** 增加连接步长 */
	private int addStep=5;
	/** 超时 时间(2个小时) */
	private long overtime=2*60*1000;
	/** 数据库类别 */
	private String dbtype;
	/**连接池*/
	private Vector<ConnectionPool> connections;
	/** 日志记录器*/
	private Logger log=null;
	
	public String getDbtype() {
		return dbtype;
	}
	public void setDbtype(String dbtype) {
		this.dbtype = dbtype;
	}

	/**
	 * 初始化数据源
	 */
	private void initDataSource() {
		log=LogUtil.getInstance(KDataSource.class);
		init();
		try {
			initConnections();
		} catch (Exception e) {
			log.error("初始化连接池错误", e);
			throw new RuntimeException("初始化连接池错误");
		}
	}
	/**
	 * 
	 */
	public KDataSource(){
		initDataSource();
	}
	/**
	 * 初始化
	 */
	private void init(){
		Properties prop=new Properties();
		try {
			InputStream input=KDataSource.class.getClassLoader().getResourceAsStream("dataSource.properties");
			prop.load(input);
			driver=prop.getProperty("driver");
			
			if(vidateString(driver)){
				throw new RuntimeException("The Driver class name is error");
			}
			Class.forName(driver.trim());//加载驱动
			url=prop.getProperty("url");
			if(vidateString(url)){
				throw new RuntimeException("The url class name is error");
			}
			url=url.trim();
			user=prop.getProperty("user");
			
			if(vidateString(user)){
				user="";
			}
			user=user.trim();
			password=prop.getProperty("password");
			if(vidateString(password)){
				password="";
			}
			password=password.trim();
			
			String itCount=prop.getProperty("initCount");
			if(vidateString(itCount)){
				initCount=10;
			}else{
				initCount=vidateInteger(itCount,10);
				if(initCount<=0){
					initCount=10;
					log.warn("The  initCount is error,you can get a default value");
				}
			}
			
			String mCount=prop.getProperty("maxCount");
			if(vidateString(mCount)){
				maxCount=50;
			}else{
				maxCount=vidateInteger(mCount,50);
				if(maxCount<=0){
					log.warn("The  maxCount is error,you can get a default value");
					maxCount=50;
				}
			}
			
			String waitStr=prop.getProperty("wait");
			if(vidateString(waitStr)){
				wait=3000;
			}else{
				wait=vidateInteger(waitStr,3000);
				if(wait<=0){
					log.warn("The  waitTime is error,you can get a default value");
					wait=3000;
				}
			}
			
			dbtype=prop.getProperty("dbtype");
			if(vidateString(dbtype)){
				dbtype="";
			}
			dbtype=dbtype.trim();
			
			
			String maxActiveCount=prop.getProperty("maxActiveCount");
			if(vidateString(maxActiveCount)){
				maxActive=5;
			}else{
				maxActive=vidateInteger(maxActiveCount,5);
				if(maxActive<=0){
				    log.warn("The  maxActive is error,you can get a default value");
					maxActive=5;
				}
			}
			
			String step=prop.getProperty("step");
			if(vidateString(step)){
				addStep=5;
			}else{
				addStep=vidateInteger(step,5);
				if(addStep<=0){
					log.warn("The  Step is error,you can get a default value");
					addStep=5;
				}
			}
			
			String over=prop.getProperty("overtime");
			if(vidateString(over)){
				overtime=2*60*60*1000;
			}else{
				overtime=vidateLong(over,2)*60*60*1000;
				if(overtime<=0){
					overtime=2*60*60*1000;
					log.warn("The  overtime is error,you can get a default value");
				}
			}
			connections=new Vector<ConnectionPool>();
		} catch (Exception e) {
			log.error("装载配置文件错误", e);
			throw new RuntimeException("装载配置文件错误");
		}
		log.info("数据源初始化成功");
	}
	/**
	 * 验证数字,如果无效则提供默认值
	 * @param itCount
	 * @param value
	 */
	private int vidateInteger(String itCount,int value) {
		int temp=0;
		try {
			itCount=itCount.trim();
			temp =Integer.parseInt(itCount);
		} catch (NumberFormatException e) {
			temp=value;
			log.warn(e.getMessage());
		}
		return temp;
	}
	/**
	 * 验证长整数,如果无效则提供默认值
	 * @param itCount
	 * @param value
	 * @return
	 */
	private long vidateLong(String itCount,long value) {
		long temp=0;
		try {
			itCount=itCount.trim();
			temp =Integer.parseInt(itCount);
		} catch (NumberFormatException e) {
			temp=value;
			log.warn(e.getMessage());
		}
		return temp;
	}
	/**
	 * 验证字符有效性
	 * @param itCount
	 * @return
	 */
	private boolean vidateString(String itCount) {
		if(itCount==null||itCount.trim().equals("")){
			return true;
		}
		return false;
	}
	/**
	 * 初始化连接池中数据库连接个数
	 *
	 */
	private void initConnections()throws Exception{
		Connection conn1=createConnection("init");
		DatabaseMetaData metaData=conn1.getMetaData();
		int dataCount=metaData.getMaxConnections();
		if(dataCount>0&&maxCount>dataCount){
			maxCount=dataCount;
		}
		if(conn1!=null){
			conn1.close();
		}
		if(initCount>=maxCount){
			initCount=maxCount;
		}
		for(int i=0;i<initCount;i++){
			try {
				Connection conn=createConnection();
				ConnectionPool pool=new ConnectionPool(conn);
				pool.setBusy(false);
				pool.setTime(System.currentTimeMillis());
				connections.addElement(pool);
			} catch (SQLException e) {
				log.error("初始化连接池错误", e);
				throw new RuntimeException("初始化连接池错误");
			}
		}
	
		log.info("初始化连接个数："+initCount);
	}
	/**
	 * 第一次创建连接对象
	 * @return 
	 * @throws SQLException
	 */
	private Connection createConnection(String init) throws SQLException{
		return DriverManager.getConnection(url.trim(),user.trim(),password.trim());
	}
	/**
	 * 创建连接对象
	 * @return 创建好的连接对象
	 * @throws SQLException
	 */
	private Connection createConnection() throws SQLException{
		Connection conn=DriverManager.getConnection(url.trim(),user.trim(),password.trim());
		if(conn!=null){
			currentCount++;
		}
		return conn;
	}
	/**
	 * 当数据源中原有连接已用完时，就创建一个新的数据库连接对象到连接池中
	 * 且这个连接需要时才被创建，一旦创建就被使用
	 * @return
	 * @throws SQLException
	 */
	private ConnectionPool createConnPool() throws SQLException{
		Connection conn=createConnection();
		ConnectionPool pool=new ConnectionPool(conn);
		pool.setBusy(true);
		pool.setTime(System.currentTimeMillis());
		connections.add(pool);
		for(int i=0;i<addStep-1;i++){
			if(currentCount<maxCount){
				Connection newconn=createConnection();
				ConnectionPool newpool=new ConnectionPool(newconn);
				pool.setBusy(false);
				pool.setTime(System.currentTimeMillis());
				connections.add(newpool);
			}
		}
		log.info("创建连接池成功, 连接池中连接数目:"+currentCount);
		return pool;
	}
	/**
	 * 补偿应超时销毁的连接（保持最大活跃连接数）
	 * @return
	 * @throws SQLException
	 */
	private ConnectionPool createOverTimeConnPool() throws SQLException{
		Connection conn=createConnection();
		ConnectionPool pool=new ConnectionPool(conn);
		pool.setBusy(false);
		pool.setTime(System.currentTimeMillis());
		connections.add(pool);
		log.info("创建连接池成功,连接池中连接数目:"+currentCount);
		return pool;
	}
	/**
	 * 从连接池中得到连接对象
	 */
	public synchronized Connection getConnection() {
		if(connections==null||connections.size()<=0){
			return null;
		}
		Connection conn=null;
		try {
			//处理超时连接
			releaseInvalidConnection();
			//查找连接
			conn = findFreeConnection();
			if(conn==null){
			  if(currentCount<maxCount){
				log.info("连接池中没有可用连接,系统创建了一个连接,  连接池中连接数目:"+currentCount);
				conn=createConnPool();
				}else{
					Thread.sleep(wait);
					//等待后再次查找
					conn=findFreeConnection();
					if(conn==null){
						throw new Exception("系统没有可用连接!!!,请稍后再试");
					}
				}
			}
		} catch (Exception e) {
			log.error("获得数据库连接失败 ", e);
			e.printStackTrace();
		} 
		return conn;
	}
	/**
	 * 查找当前连接池中还有没有空闲连接
	 * @return Connection
	 */
	@SuppressWarnings("unchecked")
	private Connection findFreeConnection(){
		Connection conn=null;
		if(connections.size()>0){
			Enumeration enu=connections.elements();
			while(enu.hasMoreElements()){
				ConnectionPool pool=(ConnectionPool)enu.nextElement();
				if((!pool.isBusy())&& (System.currentTimeMillis()-pool.getTime()<overtime)){
					conn=pool;
					pool.setBusy(true);
					return pool;
				}
			}
			if(conn==null){
				log.info("连接池中没有空闲连接, 连接池中连接数目:"+currentCount);
			}
		}
		return conn;
	}
	/**
	 * 处理超时连接
	 */
	@SuppressWarnings("unchecked")
	private void releaseInvalidConnection() throws Exception{
		if(connections==null||connections.size()<=0){
			return;
		}
		Enumeration enu=connections.elements();
		int usingconn=0;
		while(enu.hasMoreElements()){
			ConnectionPool pool=(ConnectionPool)enu.nextElement();
			if((pool.isBusy())){
				usingconn++;
			}
		}
		log.info("连接池中当前正在使用连接数:"+usingconn);
		Vector<ConnectionPool> tempPool=new Vector<ConnectionPool>();
		int i=0;
		Enumeration enu1=connections.elements();
		while(enu1.hasMoreElements()){
			ConnectionPool pool=(ConnectionPool)enu1.nextElement();
				if((System.currentTimeMillis()-pool.getTime()>=overtime)&&!pool.isBusy()){
					Connection conn=pool.getConn();
					if(conn!=null){
						conn.close();
					}
					tempPool.add(pool);
					log.info("##################### 连接池中释放一个超时连接 "+"################");
					i++;
				}
		}
		connections.removeAll(tempPool);
		currentCount=currentCount-i;
		log.info("释放无用连接后，连接池中连接数目: "+currentCount);
		//补偿应超时释放的连接
		if(currentCount<maxActive){
			for(int m=currentCount;m<maxActive;m++){
				createOverTimeConnPool();
			}
		}
		log.info("补偿超时连接后，连接池中连接数目: "+currentCount);
	}
	/**
	 * 
	 * 关闭连接池对象
	 * @param conn 
	 */
	@SuppressWarnings("unchecked")
	public synchronized void closeConnection(){
		if(connections==null||connections.size()<=0){
			return;
		}
		for(ConnectionPool pool:connections){
			if(pool!=null){
				Connection conn=pool.getConn();
				if(conn!=null){
				     try{
						if(pool.isBusy()){
							Thread.sleep(60*1000*3);//线程等待三分钟后,系统不再等待,将强制关闭所有连接
						}
						conn.close();
					} catch (Exception e) {
						log.info("释放连接失败!!!");
						e.printStackTrace();
					}
				}
			}
		}
		connections=null;
	}
	/**
	 * 单独创建一个连接，不从连接池中去取
	 */
	public synchronized Connection getConnection(String username, String password)throws SQLException {
		Connection conn=new ConnectionPool(DriverManager.getConnection(url,username,password));
		return conn;
	}
	/**
	 * 如果当前没有可用连接，设置等待时间 毫秒数
	 */
	public void setLoginTimeout(int seconds) throws SQLException {
		this.wait=seconds;
	}
	/**
	 * 或得系统等待连接时间
	 */
	public int getLoginTimeout() throws SQLException {
		return wait;
	}

	public PrintWriter getLogWriter() throws SQLException {
		return null;
	}
	public void setLogWriter(PrintWriter out) throws SQLException {
		
	}


	
}

 

package com.bpsoft.blank.jdbc;

import java.sql.CallableStatement;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.SQLWarning;
import java.sql.Savepoint;
import java.sql.Statement;
import java.util.Map;

/**
 * 包装到连接池的连接对象
 * @author newapps
 */
public class ConnectionPool implements Connection{
	/**当前数据库创建连接*/
	private Connection conn;
	/**当前连接是否处于工作状态*/
	private boolean busy=false;
	/** 记录连接创建时间 */
	private long time;
	
	public long getTime() {
		return time;
	}
	public void setTime(long time) {
		this.time = time;
	}
	/**构造函数*/
	public ConnectionPool(Connection conn){
		this.conn=conn;
	}
	public void setBusy(boolean busy){
		this.busy=busy;
	}
	public boolean isBusy(){
		return busy;
	}
	public void setConn(Connection conn){
		this.conn=conn;
	}
	public Connection getConn(){
		return conn;
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#clearWarnings()
	 */
	public void clearWarnings() throws SQLException {
		conn.clearWarnings();
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#close()
	 */
	public void close() throws SQLException {
		this.busy=false;
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#commit()
	 */
	public void commit() throws SQLException {
		conn.commit();
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#createStatement()
	 */
	public Statement createStatement() throws SQLException {
		return conn.createStatement();
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#createStatement(int, int)
	 */
	public Statement createStatement(int resultSetType,
			int resultSetConcurrency) throws SQLException {
		return conn.createStatement(resultSetType, resultSetConcurrency);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#createStatement(int, int, int)
	 */
	public Statement createStatement(int resultSetType,
			int resultSetConcurrency, int resultSetHoldability)
			throws SQLException {
		return conn.createStatement(resultSetType, resultSetConcurrency, resultSetHoldability);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#getAutoCommit()
	 */
	public boolean getAutoCommit() throws SQLException {
		return conn.getAutoCommit();
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#getCatalog()
	 */
	public String getCatalog() throws SQLException {
		return conn.getCatalog();
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#getHoldability()
	 */
	public int getHoldability() throws SQLException {
		return conn.getHoldability();
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#getMetaData()
	 */
	public DatabaseMetaData getMetaData() throws SQLException {
		return conn.getMetaData();
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#getTransactionIsolation()
	 */
	public int getTransactionIsolation() throws SQLException {
		return conn.getTransactionIsolation();
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#getTypeMap()
	 */
	public Map<String, Class<?>> getTypeMap() throws SQLException {
		return conn.getTypeMap();
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#getWarnings()
	 */
	public SQLWarning getWarnings() throws SQLException {
		return conn.getWarnings();
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#isClosed()
	 */
	public boolean isClosed() throws SQLException {
		return conn.isClosed();
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#isReadOnly()
	 */
	public boolean isReadOnly() throws SQLException {
		return conn.isReadOnly();
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#nativeSQL(java.lang.String)
	 */
	public String nativeSQL(String sql) throws SQLException {
		return conn.nativeSQL(sql);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#prepareCall(java.lang.String)
	 */
	public CallableStatement prepareCall(String sql) throws SQLException {
		return conn.prepareCall(sql);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#prepareCall(java.lang.String, int, int)
	 */
	public CallableStatement prepareCall(String sql, int resultSetType,
			int resultSetConcurrency) throws SQLException {
		return conn.prepareCall(sql, resultSetType, resultSetConcurrency);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#prepareCall(java.lang.String, int, int, int)
	 */
	public CallableStatement prepareCall(String sql, int resultSetType,
			int resultSetConcurrency, int resultSetHoldability)
			throws SQLException {
		return conn.prepareCall(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#prepareStatement(java.lang.String)
	 */
	public PreparedStatement prepareStatement(String sql)
			throws SQLException {
		return conn.prepareStatement(sql);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#prepareStatement(java.lang.String, int)
	 */
	public PreparedStatement prepareStatement(String sql,
			int autoGeneratedKeys) throws SQLException {
		return conn.prepareStatement(sql, autoGeneratedKeys);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#prepareStatement(java.lang.String, int[])
	 */
	public PreparedStatement prepareStatement(String sql,
			int[] columnIndexes) throws SQLException {
		return conn.prepareStatement(sql, columnIndexes);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#prepareStatement(java.lang.String, java.lang.String[])
	 */
	public PreparedStatement prepareStatement(String sql,
			String[] columnNames) throws SQLException {
		return conn.prepareStatement(sql, columnNames);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#prepareStatement(java.lang.String, int, int)
	 */
	public PreparedStatement prepareStatement(String sql,
			int resultSetType, int resultSetConcurrency)
			throws SQLException {
		return conn.prepareStatement(sql, resultSetType, resultSetConcurrency);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#prepareStatement(java.lang.String, int, int, int)
	 */
	public PreparedStatement prepareStatement(String sql,
			int resultSetType, int resultSetConcurrency,
			int resultSetHoldability) throws SQLException {
		return conn.prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#releaseSavepoint(java.sql.Savepoint)
	 */
	public void releaseSavepoint(Savepoint savepoint) throws SQLException {
		conn.releaseSavepoint(savepoint);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#rollback()
	 */
	public void rollback() throws SQLException {
		conn.rollback();
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#rollback(java.sql.Savepoint)
	 */
	public void rollback(Savepoint savepoint) throws SQLException {
		conn.rollback(savepoint);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#setAutoCommit(boolean)
	 */
	public void setAutoCommit(boolean autoCommit) throws SQLException {
		conn.setAutoCommit(autoCommit);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#setCatalog(java.lang.String)
	 */
	public void setCatalog(String catalog) throws SQLException {
		conn.setCatalog(catalog);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#setHoldability(int)
	 */
	public void setHoldability(int holdability) throws SQLException {
		conn.setHoldability(holdability);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#setReadOnly(boolean)
	 */
	public void setReadOnly(boolean readOnly) throws SQLException {
		conn.setReadOnly(readOnly);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#setSavepoint()
	 */
	public Savepoint setSavepoint() throws SQLException {
		return conn.setSavepoint();
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#setSavepoint(java.lang.String)
	 */
	public Savepoint setSavepoint(String name) throws SQLException {
		return conn.setSavepoint(name);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#setTransactionIsolation(int)
	 */
	public void setTransactionIsolation(int level) throws SQLException {
		conn.setTransactionIsolation(level);
	}
	/*
	 * (non-Javadoc)
	 * @see java.sql.Connection#setTypeMap(java.util.Map)
	 */
	public void setTypeMap(Map<String, Class<?>> map) throws SQLException {
		conn.setTypeMap(map);
	}
}

 

#===================================================================================
#各种常用数据库驱动名称
#============mysql Driver====================
#com.mysql.jdbc.Driver
#============oracle Driver===================
#oracle.jdbc.driver.OracleDriver
#============pointbase Driver================
#com.pointbase.jdbc.jdbcUniversalDriver
#============SQL Server Driver===============
#com.microsoft.jdbc.sqlserver.SQLServerDriver
#============DB2 Driver======================
#com.ibm.db2.jdbc.app.DB2Driver
#===================================================================================
#数据库连接url格式为:"jdbc:子协议：子协议名称//主机名:端口号/数据库名?属性名=属性值&属性名=属性值"
#dbUser和dbPassword也可以通过：属性名=属性值方式传入。
#设置数据库的编码格式：useUnicode=true&characterEncoding=GBK&zeroDateTimeBehavior=convertToNull
#============mysql url==============================================================
#jdbc:mysql://<machine_name><:port>/dbname
#jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=UTF-8
#端口号:默认是 3306
#============oracle url=============================================================
#jdbc:oracle:thin:@<machine_name><:port>:dbname
#端口号:默认是 1521
#============pointbase url==========================================================
#jdbc:pointbase:server://<machine_name><:port>/dbname
#端口号:默认是 9092
#============SQL Server url=========================================================
#jdbc:microsoft:sqlserver://<machine_name><:port>;DatabaseName=<dbname>
#端口号:默认是 1433
#============DB2 url================================================================
#jdbc:db2://<machine_name><:port>/dbname
#端口号:默认是 5000
#===================================================================================
#数据库驱动
driver=com.mysql.jdbc.Driver
#数据库URL
url=jdbc:mysql://127.0.0.1:3306/blank?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
#连接数据库用户
user=root
#连接数据库密码
password=bpcrm
#初始化连接数
initCount=2
#连接池中最大连接数
maxCount=200
#可以保留最大有效连接数
maxActiveCount=5
#最大等待时间(毫秒)
wait=3000
#===================================
#mysql数据库:'MYSQL',sqlserver数据库:'MSSQL',oracle数据库:'ORACLE'
dbtype=MYSQL
#当连接数目不够，增加连接步长
step=5
#连接超时时间(不超过4小时)-----在mysql连接超过4个小时就会失效
#overtime=2