Tomcat JDBC pool源碼部析

 最近跟隨Tomcat7.0開發了一個JDBC 鏈接池。

 Svn: http://svn.apache.org/repos/asf/tomcat/trunk/modules/jdbc-pool

http://tomcat.apache.org/tomcat-7.0-doc/jdbc-pool.html上大篇幅的介紹，包括基本的使用指南。本篇從源碼的角度，分析該鏈接池的實現思路。

應用使用JDBC鏈接也，主要關注如下三個方面：

1．  獲取鏈接

2．  歸還鏈接

3．空閒鏈接關閉。

一. 獲取鏈接

ConnectionPool提供三個接口用於獲取鏈接：

public Future<Connection> getConnectionAsync() throws SQLException {

     try {

         PooledConnection pc = borrowConnection(0, null, null);

         if (pc!=null) {

             return new ConnectionFuture(pc);

         }

     }catch (SQLException x) {

         if (x.getMessage().indexOf("NoWait")<0) {

             throw x;

         }

     }

/we can only retrieve a future if the underlying queue supports it.

     if (idle instanceof FairBlockingQueue<?>) {

         Future<PooledConnection> pcf = ((FairBlockingQueue<PooledConnection>)idle).pollAsync();

         return new ConnectionFuture(pcf);

     } else if (idle instanceof MultiLockFairBlockingQueue<?>) {

             Future<PooledConnection> pcf = ((MultiLockFairBlockingQueue<PooledConnection>)idle).pollAsync();

             return new ConnectionFuture(pcf);

     } else {

         throw new SQLException("Connection pool is misconfigured, doesn't support async retrieval. Set the 'fair' property to 'true'");

     }



public Connection getConnection() throws SQLException {

     //check out a connection

     PooledConnection con = borrowConnection(-1,null,null);

     return setupConnection(con);

 }

public Connection getConnection(String username, String password)    throws SQLException {

     // check out a connection

    PooledConnection con = borrowConnection(-1, username,      password);

     return setupConnection(con);

 }

第一個方法：getConnectionAsync用於獲取一個鏈接的Feature.它用於支持以異步的方式獲取鏈接。後兩個方法不一樣之處就是傳遞了所需鏈接的用戶名與密碼。咱們這裏得點分析第三個方法.

PooledConnection con = borrowConnection(-1, username, password);

borrowConnection方法從空閒隊列中獲取一個鏈接，或新建一個鏈接。看一下源碼：

/**

 * Thread safe way to retrieve a connection from the pool

  * @param wait - time to wait, overrides the maxWait from the properties,

  * set to -1 if you wish to use maxWait, 0 if you wish no wait time.

     * @return PooledConnection

     * @throws SQLException

     */

private PooledConnection borrowConnection(int wait, String username, String password) throws SQLException {

    //若是鏈接被關閉則直接拋出異常

        if (isClosed()) {

            throw new SQLException("Connection pool closed.");

        } //end if



        //get the current time stamp

        long now = System.currentTimeMillis();

        //see if there is one available immediately

       /*從空閒隊列中獲取一個鏈接。 其實idle裏存在鏈接對象有的可能並無

綁定物理鏈接。這也是Tomcat jdbc pool的一個特別，鏈接在將要被使用時，

纔會初始化*/

PooledConnection con = idle.poll();

        while (true) {

            if (con!=null) {

                //configure the connection and return it

                /*這裏又出現一個borrowConnection的重載方法。該方法對從空閒隊列中取到的鏈接對象進行配置和驗證,稍後評述*/

                PooledConnection result = borrowConnection(now, con, username, password);

                //null should never be returned, but was in a previous impl.

                // null should never be returned這句註釋不對，根據後面的代碼

                // 來看,null是有可能發生。

                if (result!=null) return result;

            }

            //if we get here, see if we need to create one

            //this is not 100% accurate since it doesn't use a shared

            //atomic variable - a connection can become idle while we are creating

            //a new connection

            /*從上面的英文註釋咱們很明白，當執行到這裏時，惟一的多是idle隊列沒能取到鏈接對象。
 若是條件容許，咱們將建立新的鏈接.在這裏做者用了一個特別的算法，也是tomcat代碼中經常使用的，
 咱們姑且稱他佔位法(我通常這麼叫)。這個算法的特色就是先在計數器Size中佔一個位置
 （Size是原子變量。可以解決併發問題）。即size+1.而後檢查size有沒有超標。若是超標
 則減去剛纔新加的1。不然建立一個新的鏈接。不過這裏我注意的是，若是建立新鏈接時失敗，
 size也必須減1。其實與大學時的用書搶位子殊途同歸。*/

            if (size.get() < getPoolProperties().getMaxActive()) {

                //atomic duplicate check

                if (size.addAndGet(1) > getPoolProperties().getMaxActive()) {

                  //if we got here, two threads passed through the first if

                    size.decrementAndGet();

                } else {

                    //create a connection, we're below the limit

                    //後面再描述這個方法。

                    return createConnection(now, con, username, password);

                }

            } //end if

        //到這裏則表示鏈接池已滿，不能建立新的鏈接，咱們只能等待其餘線程釋放的鏈接

            //calculate wait time for this iteration

            long maxWait = wait;

 //if the passed in wait time is -1,

 //means we should use the pool property value

            if (wait==-1) {

                maxWait = (getPoolProperties().getMaxWait()<=0)?Long.MAX_VALUE:getPoolProperties().getMaxWait();

            }

           //咱們須要計算本次最多能容忍的等待。爲何要計算呢。由於咱們可能中間被假//喚醒但卻沒能拿到鏈接。

 long timetowait = Math.max(0, maxWait - (System.currentTimeMillis() - now));

            waitcount.incrementAndGet();

            try {

                //retrieve an existing connection

                con = idle.poll(timetowait, TimeUnit.MILLISECONDS);

            } catch (InterruptedException ex) {

                if (getPoolProperties().getPropagateInterruptState()) {

                    Thread.currentThread().interrupt();

                } else {

                    Thread.interrupted();

                }

                SQLException sx = new SQLException("Pool wait interrupted.");

                sx.initCause(ex);

                throw sx;

            } finally {

                waitcount.decrementAndGet();

            }

            //no wait, return one if we have one

            if (maxWait==0 && con == null) {

                throw new PoolExhaustedException("[" + Thread.currentThread().getName()+"] " +

                "NoWait: Pool empty. Unable to fetch a connection, none available["+busy.size()+" in use].");

            }

            //we didn't get a connection, let’s see if we timed out

            if (con == null) {

                …

              if ((System.currentTimeMillis() - now) >= maxWait) {

                    throw new PoolExhaustedException("[" + Thread.currentThread().getName()+"] " +

                        "Timeout: Pool empty. Unable to fetch a connection in " + (maxWait / 1000) +

                        " seconds, none available[size:"+size.get() +"; busy:"+busy.size()+"; idle:"+idle.size()+"; lastwait:"+timetowait+"].");

                } else {

                    //no timeout, lets try again

                    //若是沒有超時，咱們繼續去獲取鏈接。

                    continue;

                }

            }

        } //while

    }

waitTime表示鏈接請求者容忍等待的最大時間，超時沒有獲取到鏈接則拋出PoolExhaustedException異常。OK。

下面咱們看中間遇到的borrowConnection的重載方法：

protected PooledConnection borrowConnection(long now, PooledConnection con, String username, String password)

和

protected PooledConnection createConnection(long now, PooledConnection notUsed, String username, String password)

首先看第一個：

/**
     * Validates and configures a previously idle connection

     * @param now - timestamp

     * @param con - the connection to validate and configure

     * @return con

     * @throws SQLException if a validation error happens

     */

 protected PooledConnection borrowConnection(long now, PooledConnection con, String username, String password) throws SQLException {

        //we have a connection, lets set it up

        //flag to see if we need to nullify

        boolean setToNull = false;

        try {

            //爲當前鏈接加鎖

            con.lock();

           //驗證當前鏈接用用戶名與密碼是否符合需求

            boolean usercheck = con.checkUser(username, password);

            if (con.isReleased()) {

                return null;

            }

            //對於沒標記爲丟棄的鏈接且沒有初始化的鏈接進行初始化。

            if (!con.isDiscarded() && !con.isInitialized()) {

                //attempt to connect

                try {

                    con.connect();

                } catch (Exception x) {

                    release(con);

                    setToNull = true;

                    if (x instanceof SQLException) {

                        throw (SQLException)x;

                    } else {

                        SQLException ex  = new SQLException(x.getMessage());

                        ex.initCause(x);

                        throw ex;

                    }

                }

            }



            if (usercheck) {

                if ((!con.isDiscarded()) && con.validate(PooledConnection.VALIDATE_BORROW)) {

                    //set the timestamp

                    con.setTimestamp(now);

                    //這裏添加LogAbandoned的功能是爲了在檢測到鏈接泄露時,

 //獲取佔用該鏈接的線程棧

                    if (getPoolProperties().isLogAbandoned()) {

                        //set the stack trace for this pool

                        con.setStackTrace(getThreadDump());

                    }

               //放入busy隊列。若是不成功，則該鏈接將沒法被追蹤。(這種狀況不會出現)

                    if (!busy.offer(con)) {

                        log.debug("Connection doesn't fit into busy array, connection will not be traceable.");

                    }

                    return con;

                }

            }

            //if we reached here, that means the connection

            //is either has another principal, is discarded or validation failed.

            //we will make one more attempt

            //in order to guarantee that the thread that just acquired

            //the connection shouldn't have to poll again.

            //這裏英語描述的很清楚了。若是鏈接的用戶名不符，被丟棄或驗證失敗，

            //咱們能夠重連該鏈接，以知足需求，而不是再去獲取其餘的。

            try {

                con.reconnect();

                if (con.validate(PooledConnection.VALIDATE_INIT)) {

                    //set the timestamp

                    con.setTimestamp(now);

                    if (getPoolProperties().isLogAbandoned()) {

                        //set the stack trace for this pool

                        con.setStackTrace(getThreadDump());

                    }

                    if (!busy.offer(con)) {

                        log.debug("Connection doesn't fit into busy array, connection will not be traceable.");

                    }

                    return con;

                } else {

                    //validation failed.

                    release(con);

                    setToNull = true;

                    throw new SQLException("Failed to validate a newly established connection.");

                }

            } catch (Exception x) {

                release(con);

                setToNull = true;

                if (x instanceof SQLException) {

                    throw (SQLException)x;

                } else {

                    SQLException ex  = new SQLException(x.getMessage());

                    ex.initCause(x);

                    throw ex;

                }

            }

        } finally {

            con.unlock();

            if (setToNull) {

                con = null;

            }

        }

    }

(待續)