使用Future模式优化一个简单的数据库连接池

我们自己写一个简单的数据库连接池,能够复用数据库连接,并且能在高并发情况下正常工作

package test;

import java.util.concurrent.ConcurrentHashMap;

public class ConnectionPool {

    private ConcurrentHashMap<String, Connection> pool = new ConcurrentHashMap<String, Connection>();
    
    public Connection getConnection(String key) {
        Connection conn = null;
        if (pool.containsKey(key)) {
            conn = pool.get(key);
        } else {
            conn = createConnection();
            pool.putIfAbsent(key, conn);
        }
        return conn;
    }
    
    public Connection createConnection() {
        return new Connection();
    }
    
    class Connection {}
}

我们用了ConcurrentHashMap,这样就不必把getConnection方法置为synchronized(当然也可以用Lock),当多个线程同时调用getConnection方法时,性能大幅提升。

貌似很完美了,但是有可能导致多余连接的创建,推演一遍:

某一时刻,同时有3个线程进入getConnection方法,调用pool.containsKey(key)都返回false,然后3个线程各自都创建了连接。虽然ConcurrentHashMap的put方法只会加入其中一个,但还是生成了2个多余的连接。如果是真正的数据库连接,那会造成极大的资源浪费。

所以,我们现在的难点是:如何在多线程访问getConnection方法时,只执行一次createConnection。

结合之前Future模式的实现分析:当3个线程都要创建连接的时候,如果只有一个线程执行createConnection方法创建一个连接,其它2个线程只需要用这个连接就行了。再延伸,把createConnection方法放到一个Callable的call方法里面,然后生成FutureTask。我们只需要让一个线程执行FutureTask的run方法,其它的线程只执行get方法就好了。

上代码:

package test;

import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

public class ConnectionPool {

    private ConcurrentHashMap<String, FutureTask<Connection>> pool = new ConcurrentHashMap<String, FutureTask<Connection>>();

    public Connection getConnection(String key) throws InterruptedException, ExecutionException {
        FutureTask<Connection> connectionTask = pool.get(key);
        if (connectionTask != null) {
            return connectionTask.get();
        } else {
            Callable<Connection> callable = new Callable<Connection>() {
                @Override
                public Connection call() throws Exception {
                    return createConnection();
                }
            };
            FutureTask<Connection> newTask = new FutureTask<Connection>(callable);
            connectionTask = pool.putIfAbsent(key, newTask);
            if (connectionTask == null) {
                connectionTask = newTask;
                connectionTask.run();
            }
            return connectionTask.get();
        }
    }

    public Connection createConnection() {
        return new Connection();
    }

    class Connection {
    }
}

推演一遍:当3个线程同时进入else语句块时,各自都创建了一个FutureTask,但是ConcurrentHashMap只会加入其中一个。第一个线程执行pool.putIfAbsent方法后返回null,然后connectionTask被赋值,接着就执行run方法去创建连接,最后get。后面的线程执行pool.putIfAbsent方法不会返回null,就只会执行get方法。

在并发的环境下,通过FutureTask作为中间转换,成功实现了让某个方法只被一个线程执行。


拓展,FutureTask的get()方法是关键,下面附上关键的代码

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    return report(s);
}
private int awaitDone(boolean timed, long nanos)
        throws InterruptedException {
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    WaitNode q = null;
    boolean queued = false;
    for (;;) {
        if (Thread.interrupted()) {
            removeWaiter(q);
            throw new InterruptedException();
        }

        int s = state;
        if (s > COMPLETING) {
            if (q != null)
                q.thread = null;
            return s;
        }
        else if (s == COMPLETING) // cannot time out yet
            Thread.yield();
        else if (q == null)
            q = new WaitNode();
        else if (!queued)
            queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                 q.next = waiters, q);
        else if (timed) {
            nanos = deadline - System.nanoTime();
            if (nanos <= 0L) {
                removeWaiter(q);
                return state;
            }
            LockSupport.parkNanos(this, nanos);
        }
        else
            LockSupport.park(this);
    }
}

get方法的逻辑很简单,如果call方法的执行过程已完成,就把结果给出去;如果未完成,就将当前线程挂起等待。awaitDone方法里面死循环的逻辑,推演几遍就能弄懂;它里面挂起线程的主要创新是定义了WaitNode类,来将多个等待线程组织成队列


下面再列出Future模式中关联到的基本的接口和类:

public class FutureTask<V> implements RunnableFuture<V> {
	/** The underlying callable; nulled out after running */
    private Callable<V> callable;
}
public interface RunnableFuture<V> extends Runnable, Future<V>
public interface Future<V> {

    boolean cancel(boolean mayInterruptIfRunning);

    boolean isCancelled();

    boolean isDone();

    V get() throws InterruptedException, ExecutionException;

    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}
public interface Callable<V> {
    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
    V call() throws Exception;
}
  • FutureTask实现了RunnableFuture接口,同时包含一个Callable实例
  • RunnableFuture继承了两个顶级接口Future和Runnable
  • 归根结底,Future、Runnable、Callable三个顶级接口决定了FutureTask的功用