Java 模拟数据库连接池的实现

前面学习过等待 – 通知机制，现在我们在其基础上添加一个超时机制，模拟从连接池中获取、使用和释放连接的过程。客户端获取连接的过程被设定为等待超时模式，即如果在 1000 毫秒内无法获取到可用连接，将会返回给客户端一个 null。设定连接池的大小为 10 个，然后通过调节客户端的线程数来模拟无法获取连接的场景

由于 java.sql.Connection 只是一个接口，最终实现是由数据库驱动提供方来实现，考虑到本例只是演示，我们通过动态代理构造一个 Connection，该 Connection 的代理仅仅是在调用 commit() 方法时休眠 100 毫秒

public class ConnectionDriver {static class ConnectionHandler implements InvocationHandler {@Overridepublic Object invoke(Object proxy, Method method, Object[] args) throws Throwable {if ("commit".equals(method.getName())) {TimeUnit.MICROSECONDS.sleep(100);}return null;}}/*** 创建一个 Connection 的代理，在 commit 时休眠 100 毫秒*/public static Connection createConnection() {return (Connection) Proxy.newProxyInstance(ConnectionDriver.class.getClassLoader(),new Class<?>[]{Connection.class}, new ConnectionHandler());}}

接下来是线程池的实现。本例通过一个双向队列来维护连接，调用方需要先调用 fetchConnection(long) 方法来指定在多少毫秒内超时获取连接，当连接使用完成后，需要调用 releaseConnection(Connection) 方法将连接放回线程池

public class ConnectionPool {private final LinkedList<Connection> pool = new LinkedList<>();public ConnectionPool(int initialSize) {// 初始化连接的最大上限if (initialSize > 0) {for (int i = 0; i < initialSize; i++) {pool.addLast(ConnectionDriver.createConnection());}}}public void releaseConnection(Connection connection) {if (connection != null) {synchronized (pool) {/* 连接释放后需要进行通知* 这样其他消费者就能知道连接池已经归还了一个连接*/pool.addLast(connection);pool.notifyAll();}}}/*** 在给定毫秒时间内获取连接*/public Connection fetchConnection(long mills) throws InterruptedException {synchronized (pool) {// 完全超时if (mills < 0) {while (pool.isEmpty()) {pool.wait();}return pool.removeFirst();} else {long future = System.currentTimeMillis() + mills;long remaining = mills;while (pool.isEmpty() && remaining > 0) {pool.wait(remaining);remaining = future - System.currentTimeMillis();}Connection result = null;if (!pool.isEmpty()) {result = pool.removeFirst();}return result;}}}}

最后编写一个用于模拟客户端获取连接的示例，该示例将模拟多个线程同时从连接池获取连接，并记录总尝试获取数、获取成功数和获取失败数

public class ConnectionPoolTest {static ConnectionPool pool = new ConnectionPool(10);static CountDownLatch start = new CountDownLatch(1);static CountDownLatch end;public static void main(String[] args) throws InterruptedException {// 线程数量int threadCount = 200;end = new CountDownLatch(threadCount);int count = 20;AtomicInteger got = new AtomicInteger();AtomicInteger notGot = new AtomicInteger();for (int i = 0; i < threadCount; i++) {Thread thread = new Thread(new ConnectionRunner(count, got, notGot), "ConnectionRunnerThread");thread.start();}start.countDown();end.await();System.out.println("total invoke : " + (threadCount * count));System.out.println("got connection : " + got);System.out.println("not got connection : " + notGot);}static class ConnectionRunner implements Runnable {int count;AtomicInteger got;AtomicInteger notGot;public ConnectionRunner(int count, AtomicInteger got, AtomicInteger notGot) {this.count = count;this.got = got;this.notGot = notGot;}@Overridepublic void run() {try {start.await();} catch (Exception e) {e.printStackTrace();}while (count > 0) {try {// 从线程池中获取连接，如果 1000ms 内无法获取到，将返回 null// 分别统计获取连接的数量 got 和未获取到的数量 notGotConnection connection = pool.fetchConnection(1000);if (connection != null) {try {connection.createStatement();connection.commit();} finally {pool.releaseConnection(connection);got.incrementAndGet();}} else {notGot.incrementAndGet();}} catch (Exception e) {e.printStackTrace();} finally {count--;}}end.countDown();}}}

笔者设置线程数量为 200 时，得出结果如下