I have made a connection pool for SQL Connection objects. The purpose of this object is to provide multi threaded support for the accessing of the underlying database.
The code works by using a reflection proxy for overriding the close() method of the returned connections, so it can detect when the consumer of the API is done processing, and therefore detect when the connection can be reused.
Connections ready for reuse are put in a Queue<>, this provides a round-robin access to the connections. If there are no usable connections remaining in this queue, then a call will be made to create a new connection from a ConnectionCreator interface.
If a exception occurs while scanning for a proper connection, then these exceptions will be put in a list, and if another exception occurs while creating the new connection, these exceptions will be added to the new exception using addSuppressed.
The API exposes the following methods to the outside:
close()- Closes the poolopenConnection()- Gets a pooled connection, or creates a new one if no connection existsgetAmountOfPooledConnections()- Get the amount of connections inside the poolcloseAllConnections()- Close all connections in the pool, but doesn't mark the object as closed.
My questions:
- Did I employ double-locking correctly in
giveBackConnection(Connection). Under no situation there may be connections be stuck in queue after closure of the object. - Did I follow correct practices for suppressing the exceptions, or should suppressed exceptions be used for something else?
- Anything else you notice about the code?
Code
ConnectionPool.java
package me.ferrybig.connectionpool;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.logging.Logger;
public class ConnectionPool implements AutoCloseable {
private static final int CONNECTION_TIMEOUT = 5000;
public static final int DEFAULT_POOL_SIZE = 16;
private static final Logger LOG = Logger.getLogger(ConnectionPool.class.getName());
private final Queue<Connection> queue = new ArrayBlockingQueue<>(5);
private final ConnectionCreator creator;
private volatile boolean isGlobalClosed = false;
public ConnectionPool(ConnectionCreator creator) {
this(creator, DEFAULT_POOL_SIZE);
}
public ConnectionPool(ConnectionCreator creator, int poolSize) {
this.creator = Objects.requireNonNull(creator, "creator == null");
}
@Override
public void close() throws SQLException {
if (isGlobalClosed) {
return;
}
isGlobalClosed = true;
closeAllConnections();
}
/**
* Get the connection to the database to access all information. This method
* supports multiple threads without creating a new connectio for each
* thread.
*
* @return Connection -> The connection to the database
* @throws SQLException
*/
public Connection openConnection() throws SQLException {
if (isGlobalClosed) {
throw new SQLException("Pool closed");
}
Connection con = getOrCreateConnection();
LOG.finest("Opening pooled connection");
return (Connection) Proxy.newProxyInstance(this.getClass().getClassLoader(),
new Class<?>[]{Connection.class}, new InvocationHandler() {
boolean isClosed = false;
@Override
public Object invoke(Object proxy, Method method, Object[] args)
throws Throwable {
if (method.getName().equals("close") && method.getParameterCount() == 0) {
assert method.getReturnType() == Void.TYPE;
if (isClosed) {
return null;
}
isClosed = true;
giveBackConnection(con);
return null;
}
return method.invoke(con, args);
}
});
}
/**
* Gets a pending connection, or a new one if none left
*
* @return Gets a pending connection, or a new one if none left
* @throws SQLException If none valid left and the creation of a new one
* failed
*/
private Connection getOrCreateConnection() throws SQLException {
List<SQLException> suppressed = new ArrayList<>();
Connection con;
while ((con = queue.poll()) != null) {
try {
if (!con.isClosed() && con.isValid(CONNECTION_TIMEOUT)) {
return con;
} else {
con.close();
}
} catch (SQLException e) {
suppressed.add(e);
}
}
try {
return createNewConnection();
} catch (SQLException | RuntimeException e) {
suppressed.forEach(e::addSuppressed);
throw e;
}
}
/**
* Create a new connection
*
* @return The newly made connection
* @throws SQLException If the creation of the <code>Connection</code>
* failed
*/
private Connection createNewConnection() throws SQLException {
LOG.finer("Creating new connection");
Connection c = creator.newConnection();
if (c == null) {
throw new SQLException("creator returned null connection");
}
return c;
}
/**
* Puts a new connection back in the global pool
*
* @param orginal The connection to put back
* @throws SQLException If it didn't fit into out pool and failed closing
*/
private void giveBackConnection(Connection orginal) throws SQLException {
LOG.finest("Closing pooled connection");
if (isGlobalClosed) {
List<SQLException> suppressed = closeRemainingConnections();
try {
LOG.fine("Closing connection");
orginal.close();
} catch (SQLException | RuntimeException e) {
suppressed.forEach(e::addSuppressed);
throw e;
}
}
if (!queue.offer(orginal)) {
LOG.fine("Closing connection");
orginal.close();
} else if (isGlobalClosed) { // variable MAY have changed
closeAllConnections();
}
}
/**
*
* @return
*/
public int getAmountOfPooledConnections() {
return queue.size();
}
public void closeAllConnections() throws SQLException {
List<SQLException> suppressed = closeRemainingConnections();
if (suppressed.isEmpty()) {
return;
}
SQLException e = new SQLException("Unable to close all connections");
suppressed.forEach(e::addSuppressed);
throw e;
}
private List<SQLException> closeRemainingConnections() {
LOG.finer("Closing remaining connection");
List<SQLException> suppressed = new ArrayList<>();
Connection c;
while ((c = queue.poll()) != null) {
try {
c.close();
} catch (SQLException e) {
suppressed.add(e);
}
}
return suppressed;
}
public interface ConnectionCreator {
public Connection newConnection() throws SQLException;
}
}
Test: ConnectionPoolTest.java
package me.ferrybig.connectionpool;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Assert;
import org.junit.Test;
import org.junit.Assume;
public class ConnectionPoolTest {
@Test
public void creatorNotCalledWhenNoConnectionIsNeededTest() throws Exception {
AtomicBoolean b = new AtomicBoolean(false);
ConnectionPool con = new ConnectionPool(() -> {
b.set(true);
Assert.fail();
return null;
});
con.close();
Assert.assertEquals(0, con.getAmountOfPooledConnections());
Assert.assertFalse(b.get());
}
@Test(expected = RuntimeException.class)
public void runtimeExceptionsCorrectlyPassedFromConnectionCreatorTest()
throws Exception {
ConnectionPool con = new ConnectionPool(() -> {
throw new RuntimeException();
});
con.openConnection();
Assert.fail();
}
@Test
public void doesntCreateMoreConnectionsThanNeededTest() throws Exception {
AtomicInteger created = new AtomicInteger(0);
try (ConnectionPool con = new ConnectionPool(() -> {
created.incrementAndGet();
return (Connection) Proxy.newProxyInstance(this.getClass().getClassLoader(),
new Class<?>[]{Connection.class}, (Object proxy, Method method, Object[] args) -> {
if ("close".equals(method.getName())) {
return null;
}
if ("isClosed".equals(method.getName())) {
return false;
}
if ("isValid".equals(method.getName())) {
return true;
}
Assert.fail("Not implemented for this test");
return null;
});
})) {
Assume.assumeTrue("Test `creatorNotCalledWhenNoConnectionIsNeededTest` is broken", created.get() == 0);
Assert.assertEquals(0, con.getAmountOfPooledConnections());
con.openConnection().close();
Assert.assertEquals(1, con.getAmountOfPooledConnections());
con.openConnection().close();
Assert.assertEquals(1, con.getAmountOfPooledConnections());
Connection c1 = con.openConnection(),
c2 = con.openConnection(),
c3 = con.openConnection(),
c4 = con.openConnection();
Assert.assertEquals(0, con.getAmountOfPooledConnections());
Assert.assertEquals(4, created.get());
c1.close();
c2.close();
c3.close();
c4.close();
Assert.assertEquals(4, con.getAmountOfPooledConnections());
Assert.assertEquals(4, created.get());
con.closeAllConnections();
Assert.assertEquals(0, con.getAmountOfPooledConnections());
Assert.assertEquals(4, created.get());
con.openConnection().close();
Assert.assertEquals(1, con.getAmountOfPooledConnections());
Assert.assertEquals(5, created.get());
}
}
@Test(expected = SQLException.class)
public void cannotGetNewConnectionAfterClosureTest() throws SQLException {
ConnectionPool con = new ConnectionPool(() -> {
throw new RuntimeException();
});
con.close();
con.openConnection();
}
}
