I had to make a database class, but I wanted it to be totally thread-safe this time, and to do that I made my own Connection wrapper class to make it easier to do that.
This class will make everything tread-safe (opening/closing ResultSets, etc), and it also adds some useful features (like a properties table, database versioning, doesTableExist method, saving a ResultSet into a List so its data is access-able while the ResultSet is closed).
To make it thread-safe, each connection can only have 1 ResultSet open at a time, and it lets other threads, that are trying to open a new ResultSet, wait until the current ResultSet is closed.
Here is my code: (to run it you need a SQL lib for the database you’re connecting with and apache beanutils, used to save a ResultSet into a List so the ResultSet can be closed while the data is still access-able)
_Main.java:
[java]package _04_mysql;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
public class _Main
{
public static void main( String[] args )
{
// config connection >>
final MyConnection connection = new MyConnection( “1.00”, MyConnection.MySQL, “localhost:3306”, “jme_game_01”, “jme_game_01”, “k77” );
// config connection <<
// open connection >>
try
{
connection.open();
}
catch( Exception e )
{
e.printStackTrace();
System.exit( 1 );
}
// open connection <<
// select something from a table >>
try
{
if( connection.doesTableExist("_data") )
{
final PreparedStatement statement = connection.prepareStatement( “SELECT * FROM _data” );
Thread thread1 = new Thread()
{
public void run()
{
try
{
ResultSet set = connection.openResultSet( statement );
System.out.println( “thread1: ResultSet has been opened” );
Thread.sleep( 1500 );
}
catch( Exception e )
{
e.printStackTrace();
System.exit( 1 );
}
System.out.println( “thread1: ResultSet has been closed\n” );
connection.closeResultSet();
}
};
Thread thread2 = new Thread()
{
public void run()
{
try
{
ResultSet set = connection.openResultSet( statement );
System.out.println( “thread2: ResultSet has been opened” );
Thread.sleep( 1500 );
}
catch( Exception e )
{
e.printStackTrace();
System.exit( 1 );
}
System.out.println( “thread2: ResultSet has been closed\n” );
connection.closeResultSet();
}
};
thread1.start();
thread2.start();
thread1.join();
thread2.join();
}
}
catch( Exception e )
{
e.printStackTrace();
System.exit( 1 );
}
try
{
System.out.println( connection.getDatabaseDataValue(“version”)+"\n" );
System.out.println( connection.getDatabaseDataValue(“test 01”) );
connection.setDatabaseDataValue( “test 01”, “test 01 – “+Math.random() );
System.out.println( connection.getDatabaseDataValue(“test 01”) );
}
catch( Exception e )
{
e.printStackTrace();
System.exit( 1 );
}
// select something from a table <<
// close connection >>
connection.close();
// close connection <<
}
}
[/java]
MyConnection.java
[java]package _04_mysql;
import java.sql.CallableStatement;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLWarning;
import java.sql.Savepoint;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.commons.beanutils.RowSetDynaClass;
public class MyConnection
{
// constants >>
private static final String databaseDataTable = “_data”; // this table will be used like a properties file
public static final Type MySQL = new Type( “mysql” );
public static final Type PostgreSQL = new Type( “postgresql” );
public static final Type H2 = new Type( “h2” );
private static final SQLException connectionIsntOpenError = new SQLException( “The connection isn’t open.” );
private static final SQLException databaseVersionLower = new SQLException( “The database is outdated.” );
private static final SQLException databaseVersionHigher = new SQLException( “The database has a higher version than this application.” );
// constants <<
// constructor parameters >>
final private String databaseVersion;
final private Type type;
final private String url;
final private String database;
final private String username;
final private String password;
// constructor parameters <<
// variables >>
private Connection connection;
private final ConcurrentLinkedQueue<Thread> openResultSetQueue = new ConcurrentLinkedQueue<Thread>();
private ResultSet currentResultSet;
private Thread currentResultSetThread;
// variables <<
public MyConnection( String databaseVersion, Type type, String url, String database, String username, String password )
{
this.databaseVersion = databaseVersion;
this.type = type;
this.url = url;
this.database = database;
this.username = username;
this.password = password;
}
public void open() throws ClassNotFoundException, InstantiationException, IllegalAccessException, SQLException
{
// open connection >>
String className = type.getJdbcClassName();
Class.forName( “com.”+className+”.jdbc.Driver” );
connection = DriverManager.getConnection( “jdbc:”+className+"://"+url+"/"+database, username, password );
// open connection <<
// prepare all statements >>
prepareAllStatements();
// prepare all statements <<
// check if database version is correct >>
String version = getDatabaseDataValue( “version” );
if( version == null )
{
setDatabaseDataValue( “version”, databaseVersion );
}
else if( !version.equals(databaseVersion) )
{
int comparedValue = version.compareTo( databaseVersion );
if( comparedValue < 0 )
throw databaseVersionLower;
else if( comparedValue > 0 )
throw databaseVersionHigher;
}
// check if database version is correct <<
}
// all prepared statements >>
//<editor-fold defaultstate=“collapsed” desc=“all prepared statements”>
/** /
private PreparedStatement statementCreateDatabaseDataTable;
/* key /
private PreparedStatement statementGetDatabaseDataValue;
/* key, value, key /
private PreparedStatement statementSetDatabaseDataValue_Insert;
/* value, key /
private PreparedStatement statementSetDatabaseDataValue_Update;
/* table /
private PreparedStatement statementDoesTableExist;
private void prepareAllStatements() throws SQLException, IllegalAccessException
{
if( statementCreateDatabaseDataTable == null )
{
String query = " CREATE TABLE “+databaseDataTable +
" (” +
" name VARCHAR( 255 ) NOT NULL PRIMARY KEY," +
" value VARCHAR( 255 ) NOT NULL" +
" );";
statementCreateDatabaseDataTable = prepareStatement( query );
}
if( statementGetDatabaseDataValue == null )
{
String query = " SELECT * " +
" FROM “+databaseDataTable +
" WHERE name = ?;”;
statementGetDatabaseDataValue = prepareStatement( query );
}
if( statementSetDatabaseDataValue_Insert == null )
{
String query = " INSERT INTO “+databaseDataTable+”( name, value )" +
" SELECT ?, ?" +
" FROM DUAL" +
" WHERE NOT EXISTS (SELECT * FROM “+databaseDataTable+” WHERE name = ?);";
statementSetDatabaseDataValue_Insert = prepareStatement( query );
}
if( statementSetDatabaseDataValue_Update == null )
{
String query = " UPDATE “+databaseDataTable +
" SET value = ?” +
" WHERE name = ?;";
statementSetDatabaseDataValue_Update = prepareStatement( query );
}
if( statementDoesTableExist == null )
{
String query = " SELECT COUNT() AS ‘count’" +
" FROM INFORMATION_SCHEMA.TABLES" +
" WHERE TABLE_SCHEMA = ‘"+database+"’"+
" AND TABLE_NAME = ?;";
statementDoesTableExist = prepareStatement( query );
}
}
//</editor-fold>
// all prepared statements <<
// queries >>
/** @return An Iterator for BasicDynaBean rows /
public Iterator openResultSetAndClose( String statement ) throws SQLException, IllegalAccessException
{
if( connection == null )
throw connectionIsntOpenError;
return openResultSetAndClose( connection.prepareStatement(statement) );
}
/* @return An Iterator for BasicDynaBean rows /
public Iterator openResultSetAndClose( PreparedStatement statement ) throws SQLException, IllegalAccessException
{
Iterator iterator = new RowSetDynaClass( openResultSet(“SELECT * FROM data”) ).getRows().iterator();
closeResultSet();
return iterator;
}
public ResultSet openResultSet( String statement ) throws SQLException, IllegalAccessException
{
if( connection == null )
throw connectionIsntOpenError;
return openResultSet( connection.prepareStatement(statement) );
}
public ResultSet openResultSet( PreparedStatement statement ) throws SQLException, IllegalAccessException
{
// wait until previous threads are done >>
Thread thread = Thread.currentThread();
synchronized( openResultSetQueue )
{
if( openResultSetQueue.contains(thread) || (thread == currentResultSetThread) )
throw new IllegalAccessException(“This thread is already opening a ResultSet, you must close the ResultSet before opening a new one.”);
openResultSetQueue.add( thread );
}
while( (currentResultSet != null) || (openResultSetQueue.peek() != thread) )
{
synchronized( thread )
{
try
{
thread.wait();
}
catch( Exception e )
{
}
}
}
// wait until previous threads are done <<
// open result set >>
synchronized( openResultSetQueue )
{
currentResultSet = statement.executeQuery();
currentResultSetThread = thread;
openResultSetQueue.poll();
}
return currentResultSet;
// open result set <<
}
public void closeResultSet()
{
synchronized( openResultSetQueue )
{
// close result set >>
try
{
if( currentResultSet != null )
if( !currentResultSet.isClosed() )
currentResultSet.close();
}
catch( SQLException e )
{
}
currentResultSetThread = null;
currentResultSet = null;
// close result set <<
// notify next thread >>
if( !openResultSetQueue.isEmpty() )
{
synchronized( openResultSetQueue.peek() )
{
openResultSetQueue.peek().notify();
}
}
// notify next thread <<
}
}
public int execute( String statement ) throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
return execute( connection.prepareStatement(statement) );
}
public int execute( PreparedStatement statement ) throws SQLException
{
return statement.executeUpdate();
}
/* calls openResultSet() /
public String getDatabaseDataValue( String key ) throws SQLException, IllegalAccessException
{
synchronized( statementCreateDatabaseDataTable )
{
if( !doesTableExist(databaseDataTable) )
return null;
}
String value = null;
synchronized( statementGetDatabaseDataValue )
{
statementGetDatabaseDataValue.setString( 1, key );
ResultSet result = openResultSet( statementGetDatabaseDataValue );
try
{
while( result.next() )
{
value = result.getString( “value” );
break;
}
}
catch( SQLException e )
{
closeResultSet();
throw e;
}
closeResultSet();
}
return value;
}
/* calls openResultSet() /
public void setDatabaseDataValue( String key, String value ) throws SQLException, IllegalAccessException
{
synchronized( statementCreateDatabaseDataTable )
{
if( !doesTableExist(databaseDataTable) )
execute( statementCreateDatabaseDataTable );
}
synchronized( statementSetDatabaseDataValue_Insert )
{
statementSetDatabaseDataValue_Insert.setString( 1, key );
statementSetDatabaseDataValue_Insert.setString( 2, value );
statementSetDatabaseDataValue_Insert.setString( 3, key );
execute( statementSetDatabaseDataValue_Insert );
}
synchronized( statementSetDatabaseDataValue_Update )
{
statementSetDatabaseDataValue_Update.setString( 1, value );
statementSetDatabaseDataValue_Update.setString( 2, key );
execute( statementSetDatabaseDataValue_Update );
}
}
/* calls openResultSet() */
public boolean doesTableExist( String table ) throws SQLException, IllegalAccessException
{
boolean value = false;
synchronized( statementDoesTableExist )
{
statementDoesTableExist.setString( 1, table );
ResultSet result = openResultSet( statementDoesTableExist );
try
{
while( result.next() )
{
value = (result.getInt(“count”) >= 1);
break;
}
}
catch( SQLException e )
{
closeResultSet();
throw e;
}
closeResultSet();
}
return value;
}
// queries <<
// default connection methods >>
public CallableStatement prepareCall( String statement ) throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
return connection.prepareCall( statement );
}
public PreparedStatement prepareStatement( String statement ) throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
return connection.prepareStatement( statement );
}
public void commit() throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
connection.commit();
}
public void rollback() throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
connection.rollback();
}
public void rollback( Savepoint savepoint ) throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
connection.rollback( savepoint );
}
public Savepoint setSavepoint() throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
return connection.setSavepoint();
}
public Savepoint setSavepoint( String name ) throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
return connection.setSavepoint( name );
}
public void releaseSavepoint( Savepoint savepoint ) throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
connection.releaseSavepoint( savepoint );
}
public void close()
{
while( (currentResultSet != null) || !openResultSetQueue.isEmpty() )
{
try
{
Thread.sleep( 50 );
}
catch( Exception e )
{
}
}
try
{
if( connection != null )
{
connection.close();
connection = null;
}
}
catch( SQLException e )
{
}
}
public boolean isClosed()
{
try
{
if( connection == null )
return true;
return connection.isClosed();
}
catch( SQLException e )
{
return true;
}
}
public SQLWarning getWarnings() throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
return connection.getWarnings();
}
public void clearWarnings() throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
connection.clearWarnings();
}
// default connection methods <<
// private classes >>
public static class Type
{
private String jdbcClassName;
private Type( String jdbcClassName )
{
this.jdbcClassName = jdbcClassName;
}
public String getJdbcClassName()
{
return jdbcClassName;
}
}
// private classes <<
}
[/java]
Please let me hear what you think of it, and let me hear it if you’ve found any possible improvements.
I’m trying to understand what this is even for. You generally DON’T want to make all database access single threaded. A good deal of a database’s code is designed so that you can access the database from multiple threads and you are essentially throwing all of that out the window to make your code run slower.
As many threads as you want can access the database as long as they have their own connection. I’m not even sure what (non JME) tutorials to direct you to on this subject.
Mysql has its own concurrency, threading protection, etc. It even has deadlock detection and locks things down potentially to a row level (depending on storage model used), etc and you can use transactions for making multiple operations atomic. You can even take read locks on rows while a transaction is processing!
I’m with pspeed here in that I’m not sure what you are trying to achieve by putting another level of thread safety over an already-concurrency-protected system.
Just create one connection per thread, don’t share them between threads, and job done…
you are using PreparedStatements which is good against SQL injection attacks, so I gave a +1
Just create one connection per thread, don’t share them between threads, and job done…
exactly, mysql server do threads for you, so you don't need ;)
Aah, could also just create a connection per thread yes, damn 
About MySQL, I wanted it to be thread-safe for all SQL databasemanagementsystems, so I only used standard SQL too.
@patrickvane1993 said:
Aah, could also just create a connection per thread yes, damn :P
About MySQL, I wanted it to be thread-safe for all SQL databasemanagementsystems, so I only used standard SQL too.
:)
It's part of the SQL contract (although it does vary between platforms) that it have thread safety in some form. Generally the "one connection per thread" model is the one used by all SQL flavours I've ever used though.
Well actually that's not quite true. Generally people use a connection pool and then get connections from and return them to the pool as they are needed, that way connections get re-used but each connection is only ever used from one thread at a time.
Plenty of good generic connection pool implements out there. Even some nice java open source ones.
Hmmm a connection pool would be useful, but it isn’t that hard to create one is it?
Just create a method grabConnection() and a method releaseConnection( con ), and an array where connections are saved in. Then let grabConnection() remove a connection from the array (or create a new one when the array is empty) and then let it return that connection, and let releaseConnection( con ) put that connection back into the array?
Why not use some ORM framework like Hibernate or just JavaEE JPA implementations like EclipseLink?
They are all thread safe, DB indipendent and very easy to use.
The connection pool is still auto managed and all you need to do is a simple xml configuration file where is specified the driver for the db.
Can those be a solution to your problems?
Yes, at it’s most basic. Most connection pools have logic about how many free connections to keep, when to let connections expire, etc. Pool management stuff.
I’ve not been thrilled by Hibernate. It’s got some nice ideas but performance can be less then stellar. It does do some stuff really well though.
This is the connectionpool that I just made:
[java]package _04_mysql;
import java.sql.SQLException;
import java.util.concurrent.ConcurrentLinkedQueue;
public class MyConnectionPool
{
private final ConcurrentLinkedQueue<MyConnection> pool = new ConcurrentLinkedQueue<MyConnection>();
private final String databaseVersion;
private final String driver;
private final String url;
private final String database;
private final String username;
private final String password;
public MyConnectionPool( String databaseVersion, String driver, String url, String database, String username, String password )
{
this.databaseVersion = databaseVersion;
this.driver = driver;
this.url = url;
this.database = database;
this.username = username;
this.password = password;
}
public MyConnection grabTempConnection() throws ClassNotFoundException, InstantiationException, IllegalAccessException, SQLException
{
synchronized( pool )
{
MyConnection connection = grabConnection();
releaseConnection( connection );
return connection;
}
}
public MyConnection grabConnection() throws ClassNotFoundException, InstantiationException, IllegalAccessException, SQLException
{
synchronized( pool )
{
if( pool.isEmpty() )
{
MyConnection connection = new MyConnection( databaseVersion, driver, url, database, username, password );
connection.open();
return connection;
}
else
{
return pool.poll();
}
}
}
public void releaseConnection( MyConnection connection )
{
synchronized( pool )
{
if( connection != null )
{
connection.closeResultSet();
pool.offer( connection );
}
}
}
}
[/java]
My Main.java changed to this:
[java]package _04_mysql;
import java.sql.ResultSet;
public class _Main
{
public static void main( String[] args )
{
// config connection >>
final MyConnectionPool connectionPool = new MyConnectionPool( "1.00", "mysql", "localhost:3306", "jme_game_01", "jme_game_01", "k77" );
// config connection <<
// test the connection >>
try
{
connectionPool.grabTempConnection();
}
catch( Exception e )
{
e.printStackTrace();
System.exit( 1 );
}
// test the connection <<
// select something from a table >>
try
{
if( connectionPool.grabTempConnection().doesTableExist("_data") )
{
final String statement = "SELECT * FROM _data";
Thread thread1 = new Thread()
{
public void run()
{
MyConnection connection = null;
try
{
connection = connectionPool.grabConnection();
ResultSet set = connection.openResultSet( statement );
System.out.println( "thread1: ResultSet has been opened" );
Thread.sleep( 1500 );
}
catch( Exception e )
{
e.printStackTrace();
System.exit( 1 );
}
System.out.println( "thread1: ResultSet has been closed" );
connection.closeResultSet();
connectionPool.releaseConnection( connection );
}
};
Thread thread2 = new Thread()
{
public void run()
{
MyConnection connection = null;
try
{
connection = connectionPool.grabConnection();
ResultSet set = connection.openResultSet( statement );
System.out.println( "thread2: ResultSet has been opened" );
Thread.sleep( 1500 );
}
catch( Exception e )
{
e.printStackTrace();
System.exit( 1 );
}
System.out.println( "thread2: ResultSet has been closed" );
connection.closeResultSet();
connectionPool.releaseConnection( connection );
}
};
thread1.start();
thread2.start();
thread1.join();
thread2.join();
}
}
catch( Exception e )
{
e.printStackTrace();
System.exit( 1 );
}
try
{
MyConnection connection = connectionPool.grabConnection();
System.out.println( connection.getDatabaseDataValue("version") );
System.out.println( connection.getDatabaseDataValue("test 01") );
connection.setDatabaseDataValue( "test 01", "test 01 – "+Math.random() );
System.out.println( connection.getDatabaseDataValue("test 01") );
connectionPool.releaseConnection( connection );
}
catch( Exception e )
{
e.printStackTrace();
System.exit( 1 );
}
// select something from a table <<
}
}
[/java]
That should work, honestly you would be better off looking at the open source ones out there though. No point re-inventing the wheel and they can put in a lot of handling (or at least detection) of corner cases that you might miss…
For example if for some reason you had 100 simultaneous databases accesses but then never had more than 2 at once for the rest of the app life cycle you would open 100 connections and then leave them open hogging resources at both ends.
Hmmm true, but that can only be solved by setting a maximum connection count? In that case I’ll add a maximum connection count to my pool as well
My connectionpool with maximum connections parameter:
[java]package _04_sql_connections;
import java.sql.SQLException;
import java.util.concurrent.ConcurrentLinkedQueue;
public class MyConnectionPool
{
private final ConcurrentLinkedQueue<MyConnection> pool = new ConcurrentLinkedQueue<MyConnection>();
private final ConcurrentLinkedQueue<Thread> grabConnectionQueue = new ConcurrentLinkedQueue<Thread>();
private final int maxConnections;
private final String databaseVersion;
private final String driver;
private final String url;
private final String database;
private final String username;
private final String password;
private int currentConnections = 0;
public MyConnectionPool( int maxConnections, String databaseVersion, String driver, String url, String database, String username, String password )
{
this.maxConnections = maxConnections;
this.databaseVersion = databaseVersion;
this.driver = driver;
this.url = url;
this.database = database;
this.username = username;
this.password = password;
}
public MyConnection grabTempConnection() throws ClassNotFoundException, InstantiationException, IllegalAccessException, SQLException
{
synchronized( pool )
{
MyConnection connection = grabConnection();
releaseConnection( connection );
return connection;
}
}
public MyConnection grabConnection() throws ClassNotFoundException, InstantiationException, IllegalAccessException, SQLException
{
synchronized( pool )
{
if( pool.isEmpty() && ((currentConnections < maxConnections) || (maxConnections <= 0)) )
{
currentConnections++;
MyConnection connection = new MyConnection( databaseVersion, driver, url, database, username, password );
connection.open();
return connection;
}
}
Thread thread = Thread.currentThread();
synchronized( grabConnectionQueue )
{
grabConnectionQueue.add( thread );
}
// wait until a connection becomes available >>
while( pool.isEmpty() || (grabConnectionQueue.peek() != thread) )
{
synchronized( thread )
{
try
{
thread.wait();
}
catch( Exception e )
{
}
}
}
// wait until previous threads are done <<
synchronized( grabConnectionQueue )
{
MyConnection connection = pool.poll();
grabConnectionQueue.poll();
return connection;
}
}
public void releaseConnection( MyConnection connection )
{
if( connection != null )
{
synchronized( pool )
{
if( !pool.contains(connection) )
{
connection.closeResultSet();
pool.add( connection );
synchronized( grabConnectionQueue )
{
if( !grabConnectionQueue.isEmpty() )
{
synchronized( grabConnectionQueue.peek() )
{
grabConnectionQueue.peek().notify();
}
}
}
}
}
}
}
}
[/java]
Now I’ll have to add a timeout per connection parameter and then it’s good enough for me.
Or by closing unused connections after a timeout period. (For example pool min 2, max 20, keepAlive 1 minute would always keep 2 connections open, other than that it would close unused connections after a minute).
What you have should work but as I already said you’re reinventing the wheel. A lot of connection pools are written by people with a lot of database knowledge who know what the best way to go about things is.
Good luck, however you go about it 
Thanks, didn’t thought about min-connections yet, which is also a good idea.
About the already made connection pools: I know people have made better pools than that I’m currently making, but I rather make something myself so I can learn something from it.
Anyway, to be continued, I think my pool will be finished in less than an hour or so.
@patrickvane1993 said:
About the already made connection pools: I know people have made better pools than that I'm currently making, but I rather make something myself so I can learn something from it.
Good mentality!
@patrickvane1993 said:
Anyway, to be continued, I think my pool will be finished in less than an hour or so.
Bad mentality!
Have a look at DBCP, BoneCP, C3P0, and the countless variants and derivatives included in the JavaEE stacks.
http://stackoverflow.com/search?q=jdbc+connection+pool+performance
Everything is done, bugtested, and ehm… done.
I’ll post the code here, in case someone sees anything that could be improved, or if someone wants to use it himself:
_Main.java
[java]package _04_sql_connections;
import java.sql.ResultSet;
public class _Main
{
public static void main( String[] args )
{
// config connection >>
final MyConnectionPool connectionPool = new MyConnectionPool( 1, 10, 60, “1.00”, “mysql”, “localhost:3306”, “jme_game_01”, “jme_game_01”, “k77” );
// config connection <<
// test a connection >>
try
{
connectionPool.grabTempConnection();
}
catch( Exception e )
{
e.printStackTrace();
System.exit( 1 );
}
// test a connection <<
// select something from a table >>
try
{
if( connectionPool.grabTempConnection().doesTableExist("_data") )
{
final String statement = “SELECT * FROM _data”;
Thread thread1 = new Thread( “thread1” )
{
public void run()
{
MyConnection connection = null;
try
{
connection = connectionPool.grabConnection();
ResultSet set = connection.openResultSet( statement );
System.out.println( “thread1: ResultSet has been opened” );
Thread.sleep( 1500 );
}
catch( Exception e )
{
e.printStackTrace();
System.exit( 1 );
}
System.out.println( “thread1: ResultSet has been closed” );
connection.closeResultSet();
connectionPool.releaseConnection( connection );
}
};
Thread thread2 = new Thread( “thread2” )
{
public void run()
{
MyConnection connection = null;
try
{
connection = connectionPool.grabConnection();
ResultSet set = connection.openResultSet( statement );
System.out.println( “thread2: ResultSet has been opened” );
Thread.sleep( 1500 );
}
catch( Exception e )
{
e.printStackTrace();
System.exit( 1 );
}
System.out.println( “thread2: ResultSet has been closed” );
connection.closeResultSet();
connectionPool.releaseConnection( connection );
}
};
thread1.start();
thread2.start();
thread1.join();
thread2.join();
}
}
catch( Exception e )
{
e.printStackTrace();
System.exit( 1 );
}
try
{
MyConnection connection = connectionPool.grabConnection();
System.out.println( connection.getDatabaseDataValue(“version”) );
System.out.println( connection.getDatabaseDataValue(“test 01”) );
connection.setDatabaseDataValue( “test 01”, “test 01 – “+Math.random() );
System.out.println( connection.getDatabaseDataValue(“test 01”) );
connectionPool.releaseConnection( connection );
}
catch( Exception e )
{
e.printStackTrace();
System.exit( 1 );
}
// select something from a table <<
}
}
[/java]
MyConnection.java
[java]package _04_sql_connections;
import java.sql.CallableStatement;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLWarning;
import java.sql.Savepoint;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.commons.beanutils.RowSetDynaClass;
public class MyConnection
{
// constants >>
private static final String databaseDataTable = “_data”; // this table will be used like a properties file
private static final SQLException connectionIsntOpenError = new SQLException( “The connection isn’t open.” );
private static final SQLException databaseVersionLower = new SQLException( “The database is outdated.” );
private static final SQLException databaseVersionHigher = new SQLException( “The database has a higher version than this application.” );
// constants <<
// constructor parameters >>
private final String databaseVersion;
private final String driver;
private final String url;
private final String database;
private final String username;
private final String password;
// constructor parameters <<
// variables >>
private Connection connection;
private final ConcurrentLinkedQueue<Thread> openResultSetQueue = new ConcurrentLinkedQueue<Thread>();
private ResultSet currentResultSet;
private Thread currentResultSetThread;
// variables <<
public MyConnection( String databaseVersion, String driver, String url, String database, String username, String password )
{
this.databaseVersion = databaseVersion;
this.driver = driver;
this.url = url;
this.database = database;
this.username = username;
this.password = password;
}
public void open() throws ClassNotFoundException, InstantiationException, IllegalAccessException, SQLException
{
// open connection >>
Class.forName( “com.”+driver+”.jdbc.Driver” );
connection = DriverManager.getConnection( “jdbc:”+driver+"://"+url+"/"+database, username, password );
// open connection <<
// prepare all statements >>
prepareAllStatements();
// prepare all statements <<
// check if database version is correct >>
String version = getDatabaseDataValue( “version” );
if( version == null )
{
setDatabaseDataValue( “version”, databaseVersion );
}
else if( !version.equals(databaseVersion) )
{
int comparedValue = version.compareTo( databaseVersion );
if( comparedValue < 0 )
throw databaseVersionLower;
else if( comparedValue > 0 )
throw databaseVersionHigher;
}
// check if database version is correct <<
}
// all prepared statements >>
//<editor-fold defaultstate=“collapsed” desc=“all prepared statements”>
/** /
private PreparedStatement statementCreateDatabaseDataTable;
/* key /
private PreparedStatement statementGetDatabaseDataValue;
/* key, value, key /
private PreparedStatement statementSetDatabaseDataValue_Insert;
/* value, key /
private PreparedStatement statementSetDatabaseDataValue_Update;
/* table /
private PreparedStatement statementDoesTableExist;
private void prepareAllStatements() throws SQLException, IllegalAccessException
{
if( statementCreateDatabaseDataTable == null )
{
String query = " CREATE TABLE “+databaseDataTable +
" (” +
" name VARCHAR( 255 ) NOT NULL PRIMARY KEY," +
" value VARCHAR( 255 ) NOT NULL" +
" );";
statementCreateDatabaseDataTable = prepareStatement( query );
}
if( statementGetDatabaseDataValue == null )
{
String query = " SELECT * " +
" FROM “+databaseDataTable +
" WHERE name = ?;”;
statementGetDatabaseDataValue = prepareStatement( query );
}
if( statementSetDatabaseDataValue_Insert == null )
{
String query = " INSERT INTO “+databaseDataTable+”( name, value )" +
" SELECT ?, ?" +
" FROM DUAL" +
" WHERE NOT EXISTS (SELECT * FROM “+databaseDataTable+” WHERE name = ?);";
statementSetDatabaseDataValue_Insert = prepareStatement( query );
}
if( statementSetDatabaseDataValue_Update == null )
{
String query = " UPDATE “+databaseDataTable +
" SET value = ?” +
" WHERE name = ?;";
statementSetDatabaseDataValue_Update = prepareStatement( query );
}
if( statementDoesTableExist == null )
{
String query = " SELECT COUNT() AS ‘count’" +
" FROM INFORMATION_SCHEMA.TABLES" +
" WHERE TABLE_SCHEMA = ‘"+database+"’"+
" AND TABLE_NAME = ?;";
statementDoesTableExist = prepareStatement( query );
}
}
//</editor-fold>
// all prepared statements <<
// queries >>
/** @return An Iterator for BasicDynaBean rows /
public Iterator openResultSetAndClose( String statement ) throws SQLException, IllegalAccessException
{
if( connection == null )
throw connectionIsntOpenError;
return openResultSetAndClose( connection.prepareStatement(statement) );
}
/* @return An Iterator for BasicDynaBean rows /
public Iterator openResultSetAndClose( PreparedStatement statement ) throws SQLException, IllegalAccessException
{
Iterator iterator = new RowSetDynaClass( openResultSet(“SELECT * FROM data”) ).getRows().iterator();
closeResultSet();
return iterator;
}
public ResultSet openResultSet( String statement ) throws SQLException, IllegalAccessException
{
if( connection == null )
throw connectionIsntOpenError;
return openResultSet( connection.prepareStatement(statement) );
}
public ResultSet openResultSet( PreparedStatement statement ) throws SQLException, IllegalAccessException
{
// wait until previous threads are done >>
Thread thread = Thread.currentThread();
synchronized( openResultSetQueue )
{
if( openResultSetQueue.contains(thread) || (thread == currentResultSetThread) )
throw new IllegalAccessException(“This thread is already opening a ResultSet, you must close the ResultSet before opening a new one.”);
openResultSetQueue.add( thread );
}
while( (currentResultSet != null) || (openResultSetQueue.peek() != thread) )
{
synchronized( thread )
{
try
{
thread.wait();
}
catch( Exception e )
{
}
}
}
// wait until previous threads are done <<
// open result set >>
synchronized( openResultSetQueue )
{
currentResultSet = statement.executeQuery();
currentResultSetThread = thread;
openResultSetQueue.poll();
}
return currentResultSet;
// open result set <<
}
public void closeResultSet()
{
synchronized( openResultSetQueue )
{
// close result set >>
try
{
if( currentResultSet != null )
if( !currentResultSet.isClosed() )
currentResultSet.close();
}
catch( SQLException e )
{
}
currentResultSetThread = null;
currentResultSet = null;
// close result set <<
// notify next thread >>
synchronized( openResultSetQueue )
{
if( !openResultSetQueue.isEmpty() )
{
synchronized( openResultSetQueue.peek() )
{
openResultSetQueue.peek().notifyAll();
}
}
}
// notify next thread <<
}
}
public int execute( String statement ) throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
return execute( connection.prepareStatement(statement) );
}
public int execute( PreparedStatement statement ) throws SQLException
{
return statement.executeUpdate();
}
/* calls openResultSet() /
public String getDatabaseDataValue( String key ) throws SQLException, IllegalAccessException
{
synchronized( statementCreateDatabaseDataTable )
{
if( !doesTableExist(databaseDataTable) )
return null;
}
String value = null;
synchronized( statementGetDatabaseDataValue )
{
statementGetDatabaseDataValue.setString( 1, key );
ResultSet result = openResultSet( statementGetDatabaseDataValue );
try
{
while( result.next() )
{
value = result.getString( “value” );
break;
}
}
catch( SQLException e )
{
closeResultSet();
throw e;
}
closeResultSet();
}
return value;
}
/* calls openResultSet() /
public void setDatabaseDataValue( String key, String value ) throws SQLException, IllegalAccessException
{
synchronized( statementCreateDatabaseDataTable )
{
if( !doesTableExist(databaseDataTable) )
execute( statementCreateDatabaseDataTable );
}
synchronized( statementSetDatabaseDataValue_Insert )
{
statementSetDatabaseDataValue_Insert.setString( 1, key );
statementSetDatabaseDataValue_Insert.setString( 2, value );
statementSetDatabaseDataValue_Insert.setString( 3, key );
execute( statementSetDatabaseDataValue_Insert );
}
synchronized( statementSetDatabaseDataValue_Update )
{
statementSetDatabaseDataValue_Update.setString( 1, value );
statementSetDatabaseDataValue_Update.setString( 2, key );
execute( statementSetDatabaseDataValue_Update );
}
}
/* calls openResultSet() */
public boolean doesTableExist( String table ) throws SQLException, IllegalAccessException
{
boolean value = false;
synchronized( statementDoesTableExist )
{
statementDoesTableExist.setString( 1, table );
ResultSet result = openResultSet( statementDoesTableExist );
try
{
while( result.next() )
{
value = (result.getInt(“count”) >= 1);
break;
}
}
catch( SQLException e )
{
closeResultSet();
throw e;
}
closeResultSet();
}
return value;
}
// queries <<
// default connection methods >>
public CallableStatement prepareCall( String statement ) throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
return connection.prepareCall( statement );
}
public PreparedStatement prepareStatement( String statement ) throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
return connection.prepareStatement( statement );
}
public void commit() throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
connection.commit();
}
public void rollback() throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
connection.rollback();
}
public void rollback( Savepoint savepoint ) throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
connection.rollback( savepoint );
}
public Savepoint setSavepoint() throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
return connection.setSavepoint();
}
public Savepoint setSavepoint( String name ) throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
return connection.setSavepoint( name );
}
public void releaseSavepoint( Savepoint savepoint ) throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
connection.releaseSavepoint( savepoint );
}
public void close()
{
while( (currentResultSet != null) || !openResultSetQueue.isEmpty() )
{
try
{
Thread.sleep( 50 );
}
catch( Exception e )
{
}
}
try
{
if( connection != null )
{
connection.close();
connection = null;
}
}
catch( SQLException e )
{
}
}
public boolean isClosed()
{
try
{
if( connection == null )
return true;
return connection.isClosed();
}
catch( SQLException e )
{
return true;
}
}
public SQLWarning getWarnings() throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
return connection.getWarnings();
}
public void clearWarnings() throws SQLException
{
if( connection == null )
throw connectionIsntOpenError;
connection.clearWarnings();
}
// default connection methods <<
}
[/java]
MyConnectionPool.java
[java]package _04_sql_connections;
import java.awt.event.ActionEvent;
import java.awt.event.ActionListener;
import java.sql.SQLException;
import java.util.Stack;
import java.util.concurrent.ConcurrentLinkedQueue;
import javax.swing.Timer;
public class MyConnectionPool
{
private final Stack<MyConnectionClass> pool = new Stack<MyConnectionClass>();
private final ConcurrentLinkedQueue<Thread> grabConnectionQueue = new ConcurrentLinkedQueue<Thread>();
private final int minConnections;
private final int maxConnections;
private final int timeoutSeconds;
private final String databaseVersion;
private final String driver;
private final String url;
private final String database;
private final String username;
private final String password;
private int currentConnections = 0;
private int currentConnectionsInUse = 0;
/**
- minConnections <= 0 means no minimum connections
- maxConnections <= 0 means no maximum connections
- timeoutSeconds <= 0 means no timeout
/
public MyConnectionPool( int minConnections, int maxConnections, int timeoutSeconds, String databaseVersion, String driver, String url, String database, String username, String password )
{
this.minConnections = minConnections;
this.maxConnections = maxConnections;
this.timeoutSeconds = timeoutSeconds;
this.databaseVersion = databaseVersion;
this.driver = driver;
this.url = url;
this.database = database;
this.username = username;
this.password = password;
closeOldConnectionsTimer.start();
}
private ActionListener closeOldConnectionsActionListener = new ActionListener()
{
public void actionPerformed( ActionEvent e )
{
closeOldConnections( minConnections, timeoutSeconds );
}
};
private Timer closeOldConnectionsTimer = new Timer( 601000, closeOldConnectionsActionListener );
public MyConnection grabTempConnection() throws ClassNotFoundException, InstantiationException, IllegalAccessException, SQLException
{
synchronized( pool )
{
synchronized( grabConnectionQueue )
{
MyConnection connection = grabConnection();
releaseConnection( connection );
return connection;
}
}
}
public MyConnection grabConnection() throws ClassNotFoundException, InstantiationException, IllegalAccessException, SQLException
{
// wait until a connection becomes available (or until one can be created) >>
Thread thread = Thread.currentThread();
synchronized( grabConnectionQueue )
{
grabConnectionQueue.add( thread );
}
while( (pool.isEmpty() && (currentConnections >= maxConnections) && (maxConnections > 0)) || (grabConnectionQueue.peek() != thread) )
{
synchronized( thread )
{
try
{
thread.wait();
}
catch( Exception e )
{
}
}
}
// wait until a connection becomes available (or until one can be created) <<
synchronized( pool )
{
synchronized( grabConnectionQueue )
{
MyConnection connection = null;
if( !pool.isEmpty() )
{
connection = pool.pop().getConnection();
}
else
{
currentConnections++;
connection = new MyConnection( databaseVersion, driver, url, database, username, password );
connection.open();
}
grabConnectionQueue.poll();
if( !grabConnectionQueue.isEmpty() )
{
synchronized( grabConnectionQueue.peek() )
{
grabConnectionQueue.peek().notifyAll();
}
}
currentConnectionsInUse++;
return connection;
}
}
}
public void releaseConnection( MyConnection connection )
{
if( connection != null )
{
currentConnectionsInUse–;
synchronized( pool )
{
connection.closeResultSet();
pool.add( new MyConnectionClass(connection) );
synchronized( grabConnectionQueue )
{
if( !grabConnectionQueue.isEmpty() )
{
synchronized( grabConnectionQueue.peek() )
{
grabConnectionQueue.peek().notifyAll();
}
}
}
}
}
}
public void closeOldConnections( int keepMinimumAmount, int timeoutSeconds )
{
synchronized( pool )
{
@SuppressWarnings( “unchecked” )
Stack<MyConnectionClass> oldPool = (Stack<MyConnectionClass>) pool.clone();
pool.clear();
int size = currentConnectionsInUse;
while( !oldPool.isEmpty() )
{
MyConnectionClass myConnectionClass = oldPool.pop();
if( (size < keepMinimumAmount) || (myConnectionClass.getStandbyTime() < timeoutSeconds) || (timeoutSeconds <= 0) )
{
size++;
pool.add( myConnectionClass );
}
else
{
myConnectionClass.getConnection().close();
currentConnections–;
}
}
}
}
private class MyConnectionClass
{
private MyConnection connection;
private long lastUsedTime = System.currentTimeMillis();
public MyConnectionClass( MyConnection connection )
{
this.connection = connection;
}
public MyConnection getConnection()
{
return connection;
}
/** @return The time in seconds this connection hasn’t been used */
public long getStandbyTime()
{
return (System.currentTimeMillis() - lastUsedTime) / 1000;
}
}
}
[/java]
I’m using bonecp currently and must say its really nice so far. As for thread safty, just have some kind of transaction runnable, that does all stuff or can rollback (including the java objects) on failture. and done.