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提供三个接口用于获取连接:


  1. public Future<Connection> getConnectionAsync() throws SQLException { 
  2.  
  3.      try { 
  4.  
  5.          PooledConnection pc = borrowConnection(0nullnull); 
  6.  
  7.          if (pc!=null) { 
  8.  
  9.              return new ConnectionFuture(pc); 
  10.  
  11.          } 
  12.  
  13.      }catch (SQLException x) { 
  14.  
  15.          if (x.getMessage().indexOf("NoWait")<0) { 
  16.  
  17.              throw x; 
  18.  
  19.          } 
  20.  
  21.      } 
  22.  
  23. /we can only retrieve a future if the underlying queue supports it. 
  24.  
  25.      if (idle instanceof FairBlockingQueue<?>) { 
  26.  
  27.          Future<PooledConnection> pcf = ((FairBlockingQueue<PooledConnection>)idle).pollAsync(); 
  28.  
  29.          return new ConnectionFuture(pcf); 
  30.  
  31.      } else if (idle instanceof MultiLockFairBlockingQueue<?>) { 
  32.  
  33.              Future<PooledConnection> pcf = ((MultiLockFairBlockingQueue<PooledConnection>)idle).pollAsync(); 
  34.  
  35.              return new ConnectionFuture(pcf); 
  36.  
  37.      } else { 
  38.  
  39.          throw new SQLException("Connection pool is misconfigured, doesn't support async retrieval. Set the 'fair' property to 'true'"); 
  40.  
  41.      } 
  42.  
  43.  
  44.  
  45. public Connection getConnection() throws SQLException { 
  46.  
  47.      //check out a connection 
  48.  
  49.      PooledConnection con = borrowConnection(-1,null,null); 
  50.  
  51.      return setupConnection(con); 
  52.  
  53.  } 
  54.  
  55. public Connection getConnection(String username, String password)    throws SQLException { 
  56.  
  57.      // check out a connection 
  58.  
  59.     PooledConnection con = borrowConnection(-1, username,      password); 
  60.  
  61.      return setupConnection(con); 
  62.  
  63.  } 

第一个方法:getConnectionAsync用于获取一个连接的Feature.它用于支持以异步的方式获取连接。后两个方法不同之处就是传递了所需连接的用户名与密码。我们这里得点分析第三个方法.

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

borrowConnection方法从空闲队列中获取一个连接,或新建一个连接。看一下源码:


  1. /** 
  2.  
  3.  * Thread safe way to retrieve a connection from the pool 
  4.  
  5.   * @param wait - time to wait, overrides the maxWait from the properties, 
  6.  
  7.   * set to -1 if you wish to use maxWait, 0 if you wish no wait time. 
  8.  
  9.      * @return PooledConnection 
  10.  
  11.      * @throws SQLException 
  12.  
  13.      */ 
  14.  
  15. private PooledConnection borrowConnection(int wait, String username, String password) throws SQLException { 
  16.  
  17.     //如果连接被关闭则直接抛出异常 
  18.  
  19.         if (isClosed()) { 
  20.  
  21.             throw new SQLException("Connection pool closed."); 
  22.  
  23.         } //end if 
  24.  
  25.   
  26.  
  27.         //get the current time stamp 
  28.  
  29.         long now = System.currentTimeMillis(); 
  30.  
  31.         //see if there is one available immediately 
  32.  
  33.        /*从空闲队列中获取一个连接。 其实idle里存在连接对象有的可能并没有 
  34.  
  35. 绑定物理连接。这也是Tomcat jdbc pool的一个特别,连接在将要被使用时, 
  36.  
  37. 才会初始化*/ 
  38.  
  39. PooledConnection con = idle.poll(); 
  40.  
  41.         while (true) { 
  42.  
  43.             if (con!=null) { 
  44.  
  45.                 //configure the connection and return it 
  46.  
  47.                 /*这里又出现一个borrowConnection的重载方法。该方法对从空闲队列中取到的连接对象进行配置和验证,稍后评述*/ 
  48.  
  49.                 PooledConnection result = borrowConnection(now, con, username, password); 
  50.  
  51.                 //null should never be returned, but was in a previous impl. 
  52.  
  53.                 // null should never be returned这句注释不对,根据后面的代码 
  54.  
  55.                 // 来看,null是有可能发生。 
  56.  
  57.                 if (result!=nullreturn result; 
  58.  
  59.             } 
  60.  
  61.             //if we get here, see if we need to create one 
  62.  
  63.             //this is not 100% accurate since it doesn't use a shared 
  64.  
  65.             //atomic variable - a connection can become idle while we are creating 
  66.  
  67.             //a new connection 
  68.  
  69.             /*从上面的英文注释我们很明白,当执行到这里时,唯一的可能是idle队列没能取到连接对象。
  70. 如果条件允许,我们将创建新的连接.在这里作者用了一个特别的算法,也是tomcat代码中常用的,
  71. 我们姑且称他占位法(我一般这么叫)。这个算法的特点就是先在计数器Size中占一个位置
  72. (Size是原子变量。能够解决并发问题)。即size+1.然后检查size有没有超标。如果超标
  73. 则减去刚才新加的1。否则创建一个新的连接。不过这里我注意的是,如果创建新连接时失败,
  74. size也必须减1。其实与大学时的用书抢位子异曲同工。*/ 
  75.  
  76.             if (size.get() < getPoolProperties().getMaxActive()) { 
  77.  
  78.                 //atomic duplicate check 
  79.  
  80.                 if (size.addAndGet(1) > getPoolProperties().getMaxActive()) { 
  81.  
  82.                   //if we got here, two threads passed through the first if 
  83.  
  84.                     size.decrementAndGet(); 
  85.  
  86.                 } else { 
  87.  
  88.                     //create a connection, we're below the limit 
  89.  
  90.                     //后面再描述这个方法。 
  91.  
  92.                     return createConnection(now, con, username, password); 
  93.  
  94.                 } 
  95.  
  96.             } //end if 
  97.  
  98.         //到这里则表示连接池已满,不能创建新的连接,我们只能等待其他线程释放的连接 
  99.  
  100.             //calculate wait time for this iteration 
  101.  
  102.             long maxWait = wait; 
  103.  
  104. //if the passed in wait time is -1, 
  105.  
  106. //means we should use the pool property value 
  107.  
  108.             if (wait==-1) { 
  109.  
  110.                 maxWait = (getPoolProperties().getMaxWait()<=0)?Long.MAX_VALUE:getPoolProperties().getMaxWait(); 
  111.  
  112.             } 
  113.  
  114.            //我们需要计算本次最多能容忍的等待。为什么要计算呢。因为我们可能中间被假//唤醒但却没能拿到连接。 
  115.  
  116. long timetowait = Math.max(0, maxWait - (System.currentTimeMillis() - now)); 
  117.  
  118.             waitcount.incrementAndGet(); 
  119.  
  120.             try { 
  121.  
  122.                 //retrieve an existing connection 
  123.  
  124.                 con = idle.poll(timetowait, TimeUnit.MILLISECONDS); 
  125.  
  126.             } catch (InterruptedException ex) { 
  127.  
  128.                 if (getPoolProperties().getPropagateInterruptState()) { 
  129.  
  130.                     Thread.currentThread().interrupt(); 
  131.  
  132.                 } else { 
  133.  
  134.                     Thread.interrupted(); 
  135.  
  136.                 } 
  137.  
  138.                 SQLException sx = new SQLException("Pool wait interrupted."); 
  139.  
  140.                 sx.initCause(ex); 
  141.  
  142.                 throw sx; 
  143.  
  144.             } finally { 
  145.  
  146.                 waitcount.decrementAndGet(); 
  147.  
  148.             } 
  149.  
  150.             //no wait, return one if we have one 
  151.  
  152.             if (maxWait==0 && con == null) { 
  153.  
  154.                 throw new PoolExhaustedException("[" + Thread.currentThread().getName()+"] " + 
  155.  
  156.                 "NoWait: Pool empty. Unable to fetch a connection, none available["+busy.size()+" in use]."); 
  157.  
  158.             } 
  159.  
  160.             //we didn't get a connection, let’s see if we timed out 
  161.  
  162.             if (con == null) { 
  163.  
  164.                 … 
  165.  
  166.               if ((System.currentTimeMillis() - now) >= maxWait) { 
  167.  
  168.                     throw new PoolExhaustedException("[" + Thread.currentThread().getName()+"] " + 
  169.  
  170.                         "Timeout: Pool empty. Unable to fetch a connection in " + (maxWait / 1000) + 
  171.  
  172.                         " seconds, none available[size:"+size.get() +"; busy:"+busy.size()+"; idle:"+idle.size()+"; lastwait:"+timetowait+"]."); 
  173.  
  174.                 } else { 
  175.  
  176.                     //no timeout, lets try again 
  177.  
  178.                     //如果没有超时,我们继续去获取连接。 
  179.  
  180.                     continue
  181.  
  182.                 } 
  183.  
  184.             } 
  185.  
  186.         } //while 
  187.  
  188.     } 

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)

首先看第一个:


  1. /** 
  2.      * Validates and configures a previously idle connection 
  3.  
  4.      * @param now - timestamp 
  5.  
  6.      * @param con - the connection to validate and configure 
  7.  
  8.      * @return con 
  9.  
  10.      * @throws SQLException if a validation error happens 
  11.  
  12.      */ 
  13.  
  14.  protected PooledConnection borrowConnection(long now, PooledConnection con, String username, String password) throws SQLException { 
  15.  
  16.         //we have a connection, lets set it up 
  17.  
  18.         //flag to see if we need to nullify 
  19.  
  20.         boolean setToNull = false
  21.  
  22.         try { 
  23.  
  24.             //为当前连接加锁
  25.  
  26.             con.lock(); 
  27.  
  28.            //验证当前连接用用户名与密码是否符合需求 
  29.  
  30.             boolean usercheck = con.checkUser(username, password); 
  31.  
  32.             if (con.isReleased()) { 
  33.  
  34.                 return null
  35.  
  36.             } 
  37.  
  38.             //对于没标记为丢弃的连接且没有初始化的连接进行初始化。 
  39.  
  40.             if (!con.isDiscarded() && !con.isInitialized()) { 
  41.  
  42.                 //attempt to connect 
  43.  
  44.                 try { 
  45.  
  46.                     con.connect(); 
  47.  
  48.                 } catch (Exception x) { 
  49.  
  50.                     release(con); 
  51.  
  52.                     setToNull = true
  53.  
  54.                     if (x instanceof SQLException) { 
  55.  
  56.                         throw (SQLException)x; 
  57.  
  58.                     } else { 
  59.  
  60.                         SQLException ex  = new SQLException(x.getMessage()); 
  61.  
  62.                         ex.initCause(x); 
  63.  
  64.                         throw ex; 
  65.  
  66.                     } 
  67.  
  68.                 } 
  69.  
  70.             } 
  71.  
  72.             
  73.  
  74.             if (usercheck) { 
  75.  
  76.                 if ((!con.isDiscarded()) && con.validate(PooledConnection.VALIDATE_BORROW)) { 
  77.  
  78.                     //set the timestamp 
  79.  
  80.                     con.setTimestamp(now); 
  81.  
  82.                     //这里添加LogAbandoned的功能是为了在检测到连接泄露时, 
  83.  
  84. //获取占用该连接的线程栈 
  85.  
  86.                     if (getPoolProperties().isLogAbandoned()) { 
  87.  
  88.                         //set the stack trace for this pool 
  89.  
  90.                         con.setStackTrace(getThreadDump()); 
  91.  
  92.                     } 
  93.  
  94.                //放入busy队列。如果不成功,则该连接将无法被追踪。(这种情况不会出现) 
  95.  
  96.                     if (!busy.offer(con)) { 
  97.  
  98.                         log.debug("Connection doesn't fit into busy array, connection will not be traceable."); 
  99.  
  100.                     } 
  101.  
  102.                     return con; 
  103.  
  104.                 } 
  105.  
  106.             } 
  107.  
  108.             //if we reached here, that means the connection 
  109.  
  110.             //is either has another principal, is discarded or validation failed. 
  111.  
  112.             //we will make one more attempt 
  113.  
  114.             //in order to guarantee that the thread that just acquired 
  115.  
  116.             //the connection shouldn't have to poll again. 
  117.  
  118.             //这里英语描述的很清楚了。如果连接的用户名不符,被丢弃或验证失败, 
  119.  
  120.             //我们可以重连该连接,以满足需求,而不是再去获取其他的。 
  121.  
  122.             try { 
  123.  
  124.                 con.reconnect(); 
  125.  
  126.                 if (con.validate(PooledConnection.VALIDATE_INIT)) { 
  127.  
  128.                     //set the timestamp 
  129.  
  130.                     con.setTimestamp(now); 
  131.  
  132.                     if (getPoolProperties().isLogAbandoned()) { 
  133.  
  134.                         //set the stack trace for this pool 
  135.  
  136.                         con.setStackTrace(getThreadDump()); 
  137.  
  138.                     } 
  139.  
  140.                     if (!busy.offer(con)) { 
  141.  
  142.                         log.debug("Connection doesn't fit into busy array, connection will not be traceable."); 
  143.  
  144.                     } 
  145.  
  146.                     return con; 
  147.  
  148.                 } else { 
  149.  
  150.                     //validation failed. 
  151.  
  152.                     release(con); 
  153.  
  154.                     setToNull = true
  155.  
  156.                     throw new SQLException("Failed to validate a newly established connection."); 
  157.  
  158.                 } 
  159.  
  160.             } catch (Exception x) { 
  161.  
  162.                 release(con); 
  163.  
  164.                 setToNull = true
  165.  
  166.                 if (x instanceof SQLException) { 
  167.  
  168.                     throw (SQLException)x; 
  169.  
  170.                 } else { 
  171.  
  172.                     SQLException ex  = new SQLException(x.getMessage()); 
  173.  
  174.                     ex.initCause(x); 
  175.  
  176.                     throw ex; 
  177.  
  178.                 } 
  179.  
  180.             } 
  181.  
  182.         } finally { 
  183.  
  184.             con.unlock(); 
  185.  
  186.             if (setToNull) { 
  187.  
  188.                 con = null
  189.  
  190.             } 
  191.  
  192.         } 
  193.  
  194.     } 

(待续)

本文转自 anranran 51CTO博客,原文链接:http://blog.51cto.com/guojuanjun/1172327

上一篇:tomcat源码分析-Bootstrap操作Catalina


下一篇:python ——python 连接字符串