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(0, null, null); 
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!=null) return 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