Mybatis源码分析(十八) - 源码包分析【数据源模块】
作者：互联网
常见的数据源组件都实现了javax.sql.DataSource接口；
MyBatis不但要能集成第三方的数据源组件，自身也提供了数据源的实现；
一般情况下，数据源的初始化过程参数较多，比较复杂；
数据源模块类图
PooledConnection：使用动态代理封装了真正的数据库连接对象；
/**
* 使用动态代理封装了真正的数据库连接对象
* @author Clinton Begin
*/
//
class PooledConnection implements InvocationHandler {
private static final String CLOSE = "close";
private static final Class<?>[] IFACES = new Class<?>[] { Connection.class };
private final int hashCode;
//记录当前连接所在的数据源对象，本次连接是有这个数据源创建的，关闭后也是回到这个数据源；
private final PooledDataSource dataSource;
//真正的连接对象
private final Connection realConnection;
//连接的代理对象
private final Connection proxyConnection;
//从数据源取出来连接的时间戳
private long checkoutTimestamp;
//连接创建的的时间戳
private long createdTimestamp;
//连接最后一次使用的时间戳
private long lastUsedTimestamp;
//根据数据库url、用户名、密码生成一个hash值，唯一标识一个连接池
private int connectionTypeCode;
//连接是否有效
private boolean valid;
/*
* Constructor for SimplePooledConnection that uses the Connection and PooledDataSource passed in
*
* @param connection - the connection that is to be presented as a pooled connection
* @param dataSource - the dataSource that the connection is from
*/
public PooledConnection(Connection connection, PooledDataSource dataSource) {
this.hashCode = connection.hashCode();
this.realConnection = connection;
this.dataSource = dataSource;
this.createdTimestamp = System.currentTimeMillis();
this.lastUsedTimestamp = System.currentTimeMillis();
this.valid = true;
this.proxyConnection = (Connection) Proxy.newProxyInstance(Connection.class.getClassLoader(), IFACES, this);
}
/*
* Required for InvocationHandler implementation.
* 此方法专门用来增强数据库connect对象，使用前检查连接是否有效，关闭时对连接进行回收
*
* @param proxy - not used
* @param method - the method to be executed
* @param args - the parameters to be passed to the method
* @see java.lang.reflect.InvocationHandler#invoke(Object, java.lang.reflect.Method, Object[])
*/
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
String methodName = method.getName();
if (CLOSE.hashCode() == methodName.hashCode() && CLOSE.equals(methodName)) {//如果是调用连接的close方法，不是真正的关闭，而是回收到连接池
dataSource.pushConnection(this);//通过pooled数据源来进行回收
return null;
} else {
try {
//使用前要检查当前连接是否有效
if (!Object.class.equals(method.getDeclaringClass())) {
// issue #579 toString() should never fail
// throw an SQLException instead of a Runtime
checkConnection();//
}
return method.invoke(realConnection, args);
} catch (Throwable t) {
throw ExceptionUtil.unwrapThrowable(t);
}
}
}
private void checkConnection() throws SQLException {
if (!valid) {
throw new SQLException("Error accessing PooledConnection. Connection is invalid.");
}
}
}
PoolState：用于管理PooledConnection对象状态的组件，通过两个list分别 管理空闲状态的连接资源和活跃状态的连
接资源
/**
* PoolState：用于管理PooledConnection对象状态的组件，通过两个list分别
* 管理空闲状态的连接资源和活跃状态的连接资源
*
* @author Clinton Begin
*/
public class PoolState {
protected PooledDataSource dataSource;
//空闲的连接池资源集合
protected final List<PooledConnection> idleConnections = new ArrayList<>();
//活跃的连接池资源集合
protected final List<PooledConnection> activeConnections = new ArrayList<>();
//请求的次数
protected long requestCount = 0;
//累计的获得连接的时间
protected long accumulatedRequestTime = 0;
//累计的使用连接的时间。从连接取出到归还，算一次使用的时间；
protected long accumulatedCheckoutTime = 0;
//使用连接超时的次数
protected long claimedOverdueConnectionCount = 0;
//累计超时时间
protected long accumulatedCheckoutTimeOfOverdueConnections = 0;
//累计等待时间
protected long accumulatedWaitTime = 0;
//等待次数
protected long hadToWaitCount = 0;
//无效的连接次数
protected long badConnectionCount = 0;
}
PooledDataSource：一个简单，同步的、线程安全的数据库连接池
getConnection() 的过程
private PooledConnection popConnection(String username, String password) throws SQLException {
boolean countedWait = false;
PooledConnection conn = null;
long t = System.currentTimeMillis();//记录尝试获取连接的起始时间戳
int localBadConnectionCount = 0;//初始化获取到无效连接的次数
while (conn == null) {
synchronized (state) {//获取连接必须是同步的
if (!state.idleConnections.isEmpty()) {//检测是否有空闲连接
// Pool has available connection
//有空闲连接直接使用
conn = state.idleConnections.remove(0);
if (log.isDebugEnabled()) {
log.debug("Checked out connection " + conn.getRealHashCode() + " from pool.");
}
} else {// 没有空闲连接
if (state.activeConnections.size() < poolMaximumActiveConnections) {//判断活跃连接池中的数量是否大于最大连接数
// 没有则可创建新的连接
conn = new PooledConnection(dataSource.getConnection(), this);
if (log.isDebugEnabled()) {
log.debug("Created connection " + conn.getRealHashCode() + ".");
}
} else {// 如果已经等于最大连接数，则不能创建新连接
//获取最早创建的连接
PooledConnection oldestActiveConnection = state.activeConnections.get(0);
long longestCheckoutTime = oldestActiveConnection.getCheckoutTime();
if (longestCheckoutTime > poolMaximumCheckoutTime) {//检测是否已经以及超过最长使用时间
// 如果超时，对超时连接的信息进行统计
state.claimedOverdueConnectionCount++;//超时连接次数+1
state.accumulatedCheckoutTimeOfOverdueConnections += longestCheckoutTime;//累计超时时间增加
state.accumulatedCheckoutTime += longestCheckoutTime;//累计的使用连接的时间增加
state.activeConnections.remove(oldestActiveConnection);//从活跃队列中删除
if (!oldestActiveConnection.getRealConnection().getAutoCommit()) {//如果超时连接未提交，则手动回滚
try {
oldestActiveConnection.getRealConnection().rollback();
} catch (SQLException e) {//发生异常仅仅记录日志
/*
Just log a message for debug and continue to execute the following
statement like nothing happend.
Wrap the bad connection with a new PooledConnection, this will help
to not intterupt current executing thread and give current thread a
chance to join the next competion for another valid/good database
connection. At the end of this loop, bad {@link @conn} will be set as null.
*/
log.debug("Bad connection. Could not roll back");
}
}
//在连接池中创建新的连接，注意对于数据库来说，并没有创建新连接；
conn = new PooledConnection(oldestActiveConnection.getRealConnection(), this);
conn.setCreatedTimestamp(oldestActiveConnection.getCreatedTimestamp());
conn.setLastUsedTimestamp(oldestActiveConnection.getLastUsedTimestamp());
//让老连接失效
oldestActiveConnection.invalidate();
if (log.isDebugEnabled()) {
log.debug("Claimed overdue connection " + conn.getRealHashCode() + ".");
}
} else {
// 无空闲连接，最早创建的连接没有失效，无法创建新连接，只能阻塞
try {
if (!countedWait) {
state.hadToWaitCount++;//连接池累计等待次数加1
countedWait = true;
}
if (log.isDebugEnabled()) {
log.debug("Waiting as long as " + poolTimeToWait + " milliseconds for connection.");
}
long wt = System.currentTimeMillis();
state.wait(poolTimeToWait);//阻塞等待指定时间
state.accumulatedWaitTime += System.currentTimeMillis() - wt;//累计等待时间增加
} catch (InterruptedException e) {
break;
}
}
}
}
if (conn != null) {//获取连接成功的，要测试连接是否有效，同时更新统计数据
// ping to server and check the connection is valid or not
if (conn.isValid()) {//检测连接是否有效
if (!conn.getRealConnection().getAutoCommit()) {
conn.getRealConnection().rollback();//如果遗留历史的事务，回滚
}
//连接池相关统计信息更新
conn.setConnectionTypeCode(assembleConnectionTypeCode(dataSource.getUrl(), username, password));
conn.setCheckoutTimestamp(System.currentTimeMillis());
conn.setLastUsedTimestamp(System.currentTimeMillis());
state.activeConnections.add(conn);
state.requestCount++;
state.accumulatedRequestTime += System.currentTimeMillis() - t;
} else {//如果连接无效
if (log.isDebugEnabled()) {
log.debug("A bad connection (" + conn.getRealHashCode() + ") was returned from the pool, getting another connection.");
}
state.badConnectionCount++;//累计的获取无效连接次数+1
localBadConnectionCount++;//当前获取无效连接次数+1
conn = null;
//拿到无效连接，但如果没有超过重试的次数，允许再次尝试获取连接，否则抛出异常
if (localBadConnectionCount > (poolMaximumIdleConnections + poolMaximumLocalBadConnectionTolerance)) {
if (log.isDebugEnabled()) {
log.debug("PooledDataSource: Could not get a good connection to the database.");
}
throw new SQLException("PooledDataSource: Could not get a good connection to the database.");
}
}
}
}
}
if (conn == null) {
if (log.isDebugEnabled()) {
log.debug("PooledDataSource: Unknown severe error condition. The connection pool returned a null connection.");
}
throw new SQLException("PooledDataSource: Unknown severe error condition. The connection pool returned a null connection.");
}
return conn;
}
pushConnection()过程
//回收连接资源
protected void pushConnection(PooledConnection conn) throws SQLException {
synchronized (state) {//回收连接必须是同步的
state.activeConnections.remove(conn);//从活跃连接池中删除此连接
if (conn.isValid()) {
//判断闲置连接池资源是否已经达到上限
if (state.idleConnections.size() < poolMaximumIdleConnections && conn.getConnectionTypeCode() == expectedConnectionTypeCode) {
//没有达到上限，进行回收
state.accumulatedCheckoutTime += conn.getCheckoutTime();
if (!conn.getRealConnection().getAutoCommit()) {
conn.getRealConnection().rollback();//如果还有事务没有提交，进行回滚操作
}
//基于该连接，创建一个新的连接资源，并刷新连接状态
PooledConnection newConn = new PooledConnection(conn.getRealConnection(), this);
state.idleConnections.add(newConn);
newConn.setCreatedTimestamp(conn.getCreatedTimestamp());
newConn.setLastUsedTimestamp(conn.getLastUsedTimestamp());
//老连接失效
conn.invalidate();
if (log.isDebugEnabled()) {
log.debug("Returned connection " + newConn.getRealHashCode() + " to pool.");
}
//唤醒其他被阻塞的线程
state.notifyAll();
} else {//如果闲置连接池已经达到上限了，将连接真实关闭
state.accumulatedCheckoutTime += conn.getCheckoutTime();
if (!conn.getRealConnection().getAutoCommit()) {
conn.getRealConnection().rollback();
}
//关闭真的数据库连接
conn.getRealConnection().close();
if (log.isDebugEnabled()) {
log.debug("Closed connection " + conn.getRealHashCode() + ".");
}
//将连接对象设置为无效
conn.invalidate();
}
} else {
if (log.isDebugEnabled()) {
log.debug("A bad connection (" + conn.getRealHashCode() + ") attempted to return to the pool, discarding connection.");
}
state.badConnectionCount++;
}
}
}
标签：log,数据源,state,PooledConnection,connection,源码,Mybatis,连接,conn 来源： https://blog.csdn.net/a1173537204/article/details/94025292