摘要：池化技術(shù)簡介在我們使用數(shù)據(jù)庫的過程中，我們往往使用數(shù)據(jù)庫連接池而不是直接使用數(shù)據(jù)庫連接進(jìn)行操作，這是因?yàn)槊恳粋€(gè)數(shù)據(jù)庫連接的創(chuàng)建和銷毀的代價(jià)是昂貴的，而池化技術(shù)則預(yù)先創(chuàng)建了資源，這些資源是可復(fù)用的這樣就保證了在多用戶情況下只能使用指定數(shù)目的資

在我們使用數(shù)據(jù)庫的過程中，我們往往使用數(shù)據(jù)庫連接池而不是直接使用數(shù)據(jù)庫連接進(jìn)行操作，這是因?yàn)槊恳粋€(gè)數(shù)據(jù)庫連接的創(chuàng)建和銷毀的代價(jià)是昂貴的，而池化技術(shù)則預(yù)先創(chuàng)建了資源，這些資源是可復(fù)用的,這樣就保證了在多用戶情況下只能使用指定數(shù)目的資源，避免了一個(gè)用戶創(chuàng)建一個(gè)連接資源，造成程序運(yùn)行開銷過大。

這里只實(shí)現(xiàn)一個(gè)簡易的連接池，更多復(fù)雜的需求可根據(jù)該連接池進(jìn)行改進(jìn)，該連接池主要參數(shù)如下：

一個(gè)繁忙隊(duì)列busy

一個(gè)空閑隊(duì)列idle

連接池最大活動(dòng)連接數(shù)maxActive

連接池最大等待時(shí)間maxWait

連接池的活動(dòng)連接數(shù)activeSize

程序流程圖如下：

泛型接口ConnectionPool.java

public interface ConnectionPool {

    /**
     * 初始化池資源
     * @param maxActive 池中最大活動(dòng)連接數(shù)
     * @param maxWait 最大等待時(shí)間
     */
    void init(Integer maxActive, Long maxWait);

    /**
     * 從池中獲取資源
     * @return 連接資源
     */
    T getResource() throws Exception;

    /**
     * 釋放連接
     * @param connection 正在使用的連接
     */
    void release(T connection) throws Exception;

    /**
     * 釋放連接池資源
     */
    void close();


}

以zookeeper為例，實(shí)現(xiàn)zookeeper連接池，ZookeeperConnectionPool.java

public class ZookeeperConnectionPool implements ConnectionPool {
    //最大活動(dòng)連接數(shù)
    private Integer maxActive; 
    //最大等待時(shí)間
    private Long maxWait; 
    //空閑隊(duì)列
    private LinkedBlockingQueue idle = new LinkedBlockingQueue<>();
    //繁忙隊(duì)列
    private LinkedBlockingQueue busy = new LinkedBlockingQueue<>();
    //連接池活動(dòng)連接數(shù)
    private AtomicInteger activeSize = new AtomicInteger(0);
    //連接池關(guān)閉標(biāo)記
    private AtomicBoolean isClosed = new AtomicBoolean(false);
    //總共獲取的連接記數(shù)
    private AtomicInteger createCount = new AtomicInteger(0);
    //等待zookeeper客戶端創(chuàng)建完成的計(jì)數(shù)器
    private static ThreadLocal latchThreadLocal = ThreadLocal.withInitial(() -> new CountDownLatch(1));

    public ZookeeperConnectionPool(Integer maxActive, Long maxWait) {
        this.init(maxActive, maxWait);
    }

    @Override
    public void init(Integer maxActive, Long maxWait) {
        this.maxActive = maxActive;
        this.maxWait = maxWait;
    }

    @Override
    public ZooKeeper getResource() throws Exception {
        ZooKeeper zooKeeper;
        Long nowTime = System.currentTimeMillis();
        final CountDownLatch countDownLatch = latchThreadLocal.get();
        
        //空閑隊(duì)列idle是否有連接
        if ((zooKeeper = idle.poll()) == null) {
            //判斷池中連接數(shù)是否小于maxActive
            if (activeSize.get() < maxActive) {
                //先增加池中連接數(shù)后判斷是否小于等于maxActive
                if (activeSize.incrementAndGet() <= maxActive) {
                    //創(chuàng)建zookeeper連接
                    zooKeeper = new ZooKeeper("localhost", 5000, (watch) -> {
                        if (watch.getState() == Watcher.Event.KeeperState.SyncConnected) {
                            countDownLatch.countDown();
                        }
                    });
                    countDownLatch.await();
                    System.out.println("Thread:" + Thread.currentThread().getId() + "獲取連接：" + createCount.incrementAndGet() + "條");
                    busy.offer(zooKeeper);
                    return zooKeeper;
                } else {
                    //如增加后發(fā)現(xiàn)大于maxActive則減去增加的
                    activeSize.decrementAndGet();
                }
            }
            //若活動(dòng)線程已滿則等待busy隊(duì)列釋放連接
            try {
                System.out.println("Thread:" + Thread.currentThread().getId() + "等待獲取空閑資源");
                Long waitTime = maxWait - (System.currentTimeMillis() - nowTime);
                zooKeeper = idle.poll(waitTime, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                throw new Exception("等待異常");
            }
            //判斷是否超時(shí)
            if (zooKeeper != null) {
                System.out.println("Thread:" + Thread.currentThread().getId() + "獲取連接：" + createCount.incrementAndGet() + "條");
                busy.offer(zooKeeper);
                return zooKeeper;
            } else {
                System.out.println("Thread:" + Thread.currentThread().getId() + "獲取連接超時(shí)，請重試！");
                throw new Exception("Thread:" + Thread.currentThread().getId() + "獲取連接超時(shí)，請重試！");
            }
        }
        //空閑隊(duì)列有連接，直接返回
        busy.offer(zooKeeper);
        return zooKeeper;
    }

    @Override
    public void release(ZooKeeper connection) throws Exception {
        if (connection == null) {
            System.out.println("connection 為空");
            return;
        }
        if (busy.remove(connection)){
            idle.offer(connection);
        } else {
            activeSize.decrementAndGet();
            throw new Exception("釋放失敗");
        }
    }

    @Override
    public void close() {
        if (isClosed.compareAndSet(false, true)) {
            idle.forEach((zooKeeper) -> {
                try {
                    zooKeeper.close();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            busy.forEach((zooKeeper) -> {
                try {
                    zooKeeper.close();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
    }
}

這里創(chuàng)建20個(gè)線程并發(fā)測試連接池，Test.java

public class Test {

    public static void main(String[] args) throws Exception {
        int threadCount = 20;
        Integer maxActive = 10;
        Long maxWait = 10000L;
        ZookeeperConnectionPool pool = new ZookeeperConnectionPool(maxActive, maxWait);
        CountDownLatch countDownLatch = new CountDownLatch(20);
        for (int i = 0; i < threadCount; i++) {
            new Thread(() -> {
                countDownLatch.countDown();
                try {
                    countDownLatch.await();
                    ZooKeeper zooKeeper = pool.getResource();
                    Thread.sleep(2000);
                    pool.release(zooKeeper);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }).start();
        }
        while (true){

        }
    }
}