连接池的实现和优化(一)

    由于数据库连接，JMS连接是一种较为昂贵的资源，创建连接需要花不少时间，通常在连接创建后，会将连接缓存在连接池中，以便减少创建连接的时间和重复使用连接，节约资源和提高效率。关于连接池优点，介绍的文章较多，这儿就不再赘述了。下面主要通过示例代码来讨论如何实现一个连接池，并找出其中存在的问题，在后面的系列中对其进行优化。

    连接池，对于有用户来说，最关心的就是能从连接池中获取连接，并能在连接关闭时，自动将连接归还到连接池中。一个最简单的连接池接口抽象如下：

public interface ConnectionPool {

    /**
     * 获取连接
     */
    public Connection getConnection();

    /**
     * 释放连接
     */
    public void releaseConnection(Connection conn);

}

一个模拟测试的连接类如下：

public class Connection {

    /**连接池*/
    private final ConnectionPool pool;

    public Connection(ConnectionPool pool) {
        this.pool = pool;
    }

    /**
     * 关闭连接，将连接返回到连接池中
     */
    public void close() {
        pool.releaseConnection(this);
    }

    //省略其他逻辑方法 ... ...
}

对于连接池的实现，除了能从中获取连接和将不用的连接归还到连接池中，还必须有调整连接池大小的功能，当连接池中的空闲连接不够使用时，连接池需要创建新的连接来满足请求连接的线程，如果创建的连接已经达到连接池所限定最大数目且仍然没有连接可用时，需要将请求连接的线程加入等待队列中等待，有连接归还时，按照先进先出的原则唤醒等待线程去获取连接。下面是连接池的实现代码（由于篇幅关系没有实现超时，连接池空闲时减少连接数目等功能）：

import java.util.*;

public class ConnectionPoolImpl implements ConnectionPool {

    /**空闲连接*/
    private List<Connection> freeList = new ArrayList<Connection>();
    /**等待线程队列，先进去出*/
    private final LinkedList<Object> waitQueue = new LinkedList<Object>();
    /**最小连接数*/
    private int minSize = 0;
    /**最大连接数*/
    private int maxSize = 32;
    /**连接池中的总连接数*/
    private int totalSize = 0;
    /**连接池调整大小*/
    private int step = 1;
    /**连接池是否已经初始化*/
    private boolean initialized = false;
    /**调试模式*/
    private boolean debug = false;

    /**
     * 初始化池
     */
    private synchronized void initPool() {
        if (initialized) {
            return;
        }

        initialized = true;

        if (debug) debugPrint("Connection pool initialized!");

        for (int i = 0; i < minSize; i++) {
            Connection conn = new Connection(this);
            freeList.add(conn);
            ++totalSize;
            if (debug) {
                debugPrint("Increase a connection, " +
                        "total connections =" + totalSize
                        + ", free connections " + freeList.size());
            }
        }
    }

    /**
     * 获取连接，如果当前没有连接可用，则加入等待队列
     */
    public Connection getConnection() {

        Connection result = null;

        while (true) {//直到获取到一条连接为止
            result = internalGetConnection();

            if (result != null) {
                if (debug) {
                    debugPrint("Thread " + Thread.currentThread().getName()
                            + " aquired a connection, " +
                            "total connections =" + totalSize
                            + ", free connections " + freeList.size());
                }
                break;
            } else {
                Object monitor = new Object();
                if (debug) {
                    debugPrint("Thread " + Thread.currentThread().getName()
                            + " wait for a connection.");
                }

                //没有获取到连接，将当前线程加入等待队列
                synchronized (monitor) {
                    synchronized (waitQueue) {waitQueue.add(monitor);}
                    try {monitor.wait();} catch (InterruptedException ignore) {}
                    //唤醒后会继续回到while循环，尝试获取连接
                }

                if (debug) {
                    debugPrint("Thread " + Thread.currentThread().getName()
                            + " wakeup.");
                }
            }
        }


        return result;
    }

    /**
     * 获取连接，如果没有连接，则尝试增加连接池
     */
    private synchronized Connection internalGetConnection() {
        if (!initialized) {
            initPool();
        }

        Connection result = null;
        if (!freeList.isEmpty()) {
            result = freeList.remove(0);
        } else {
            if (totalSize < maxSize) {
                if (debug) {
                    debugPrint("Current pool is empty, " +
                            "try to increase connection pool.");
                }
                //当前创建的连接总数小于最大连接数，增加连接池
                result = increasePool();
            }
        }

        return result;
    }

    /**
     * 增加连接池，同时将最后创建的连接返回给当前线程
     */
    private Connection increasePool() {
        int localStep = step;
        if (totalSize + step > maxSize) {
            localStep = maxSize - totalSize;
        }

        Connection result = null;
        int lastIndex = localStep - 1;
        for (int i = 0; i < localStep; i++) {
            Connection conn = new Connection(this);
            ++totalSize;

            if (i == lastIndex) {
                result = conn;//最后创建的连接返回给当前线程使用
                if (debug) {
                    debugPrint("Increate a connection, " +
                            "total connections =" + totalSize
                            + ", free connections " + freeList.size());
                }
            } else {
                freeList.add(conn);
                if (debug) {
                    debugPrint("Increate a connection, "
                            + "total connections =" + totalSize
                            + ", free connections " + freeList.size());
                }
                //增加连接后唤醒等待线程
                notifyWaitingThreads();
            }
        }

        return result;
    }

    /**
     * 唤醒等待的线程
     */
    private void notifyWaitingThreads() {
        Object waitMonitor = null;
        synchronized (waitQueue) {
            if (waitQueue.size() > 0) {
                waitMonitor = waitQueue.removeFirst();
            }
        }

        if (waitMonitor != null) {
            synchronized (waitMonitor) {
                waitMonitor.notify();
            }
        }
    }

    /**
     * 释放连接，同时唤醒等待的线程
     */
    public synchronized void releaseConnection(Connection conn) {
        freeList.add(conn);
        if (debug) {
            debugPrint("Release a connection, "
                    + "total connections =" + totalSize
                    + ", free connections " + freeList.size());
        }
        notifyWaitingThreads();
    }


    private void debugPrint(String debugStr) {
        System.out.println(debugStr);
    }

    //省略其他getter, setter ...
}

下面创建一个连接池，并使用较多的线程来从连接池中不断获取连接，并使用一段时间后关闭连接。经测试，如果每个线程一次只获取一条连接并在使用完后关闭连接将其归还到连接池中，上面的连接池实现已经能支持较多并发不会出现问题。测试代码如下：

import java.util.Random;

public class ConnectionPoolTest {

    /**
     * 模拟客户端线程不断获取，释放连接的情况
     */
    private static class PoolClient implements Runnable {

        private static int idGenerator = 0;
        private final ConnectionPoolImpl pool;
        private final String threadName;
        private Random random = new Random();

        public PoolClient(ConnectionPoolImpl pool) {
            this.pool = pool;
            threadName = "pool-client-" + (++idGenerator);
            Thread.currentThread().setName(threadName);
        }

        public void run() {

            for (int i = 0; i < 100; i++) {
                Connection conn = pool.getConnection();
                System.out.println("Thread " + threadName
                        + " aquired a connection.");
                int sleepTime = (3 + random.nextInt(20)) * 1000;
                try {
                    Thread.sleep(sleepTime);
                } catch (InterruptedException ignore) {
                }
                conn.close();
                System.out.println("Thread " + threadName
                        + " release a connection.");
            }
        }
    }

    public static void main(String[] args) throws Exception {

        ConnectionPoolImpl pool = new ConnectionPoolImpl();
        pool.setMaxSize(100);
        pool.setMinSize(5);
        pool.setDebug(true);

        //测试连接池，模拟1000个线程不断获取释放连接的情况
        for (int i = 0; i < 1000; i++) {
            Thread thread = new Thread(new PoolClient(pool));
            thread.start();
        }

    }
}

   上面是一个简单的连接池的实现，该连接池已基本能满足连接管理的功能，能缓存连接，并在连接不够使用时，对请求连接的线程进行排队，如果有连接回到连接池中时，会按照先进先出的原则唤醒等待队列中的线程。但是却存在以下较明显的问题：

     
• 连接池的初始化效率较低，连接池initPool()方法占用了整个连接池的锁，如果同时有很多线程在连接池尚未初始化时并发的调用getConnection()来获取连接，那么所有的线程都得等待initPool()将连接池初始化完毕后才可能获取到连接，如果初始化的时间足够长，将导致很多线程需要等待较长时间不能工作。
     
• 连接池的调整效率低下，同连接池初始化initPool()方法一样，连接池调整increasePool()也占用了整个连接池的锁，所以在调整连接池过程中，其他线程无法获取连接和归还连接，如果连接池的调整时间足够长，由于其他线程因为无法归还连接和获取连接导致连接池效率极其低下。
    
• 该连接池基本能满足每个线程从中取回一个连接，用完后就即使归还的情况。如果有部分线程需要取两个连接或者三个连接，用完后再一起归还，使用该连接池时，很明显会出现资源竞争饥饿死锁的情况。

在后面的系列中将讲解如何优化代码避免这些情况。

评论

1 楼 hsudb 2012-05-27  

楼主真2,并排占沙发