myBatis源码解析-数据源篇（3）
- 2020 年 8 月 5 日
- 筆記
前言：我们使用mybatis时，关于数据源的配置多使用如c3p0,druid等第三方的数据源。其实mybatis内置了数据源的实现，提供了连接数据库，池的功能。在分析了缓存和日志包的源码后，接下来分析mybatis中的数据源实现。
类图：mybatis中关于数据源的源码包路径如下：
mybatis中提供了一个DataSourceFactory接口，提供了设置数据源配置信息，获取数据源方法。查看类图可知，有三个实现类分别提供了不同的数据源实现。JndiDataSourceFactory,PooledDataSourceFactory,unPooledDataSourceFactory。JndiDataSourceFactory实现较简单，此处源码略过。如下为各类的相互关系。
unPooledDataSourceFactory，PooledDataSourceFactory源码分析：unPooledDataSourceFactory实现了DataSourceFactory接口，实现了数据源配置及获取数据源方法。
// 对外提供的数据源工厂接口 public interface DataSourceFactory { // 设置配置信息 void setProperties(Properties props); // 获取数据源 DataSource getDataSource(); }
// 非池化的数据源工厂类 public class UnpooledDataSourceFactory implements DataSourceFactory { private static final String DRIVER_PROPERTY_PREFIX = "driver."; // 数据库驱动名前缀 private static final int DRIVER_PROPERTY_PREFIX_LENGTH = DRIVER_PROPERTY_PREFIX.length(); protected DataSource dataSource; // 数据源 public UnpooledDataSourceFactory() { this.dataSource = new UnpooledDataSource(); // 构造一个非池化的数据源（下文分析数据源详细代码） } public void setProperties(Properties properties) { // 对数据源进行配置,此处设计反射包的知识（本章重点不在这，可忽略） Properties driverProperties = new Properties(); MetaObject metaDataSource = SystemMetaObject.forObject(dataSource); // 将dataSource类转为metaObject类 for (Object key : properties.keySet()) { String propertyName = (String) key; if (propertyName.startsWith(DRIVER_PROPERTY_PREFIX)) { // 若是数据库驱动配置 String value = properties.getProperty(propertyName); driverProperties.setProperty(propertyName.substring(DRIVER_PROPERTY_PREFIX_LENGTH), value); // driverProperties存储数据库驱动参数 } else if (metaDataSource.hasSetter(propertyName)) { // 如果有set方法 String value = (String) properties.get(propertyName); // 根据属性类型进行类型的转换，主要是 Integer, Long, Boolean 三种类型的转换 Object convertedValue = convertValue(metaDataSource, propertyName, value); // 设置DataSource 的相关属性值 metaDataSource.setValue(propertyName, convertedValue); } else { throw new DataSourceException("Unknown DataSource property: " + propertyName); } } // 设置 DataSource.driverProerties 属性值 if (driverProperties.size() > 0) { metaDataSource.setValue("driverProperties", driverProperties); } } // 获取数据源 public DataSource getDataSource() { return dataSource; } // 对Integer, Long, Boolean 三种类型的转换 private Object convertValue(MetaObject metaDataSource, String propertyName, String value) { Object convertedValue = value; Class<?> targetType = metaDataSource.getSetterType(propertyName); if (targetType == Integer.class || targetType == int.class) { convertedValue = Integer.valueOf(value); } else if (targetType == Long.class || targetType == long.class) { convertedValue = Long.valueOf(value); } else if (targetType == Boolean.class || targetType == boolean.class) { convertedValue = Boolean.valueOf(value); } return convertedValue; } }
public class PooledDataSourceFactory extends UnpooledDataSourceFactory { public PooledDataSourceFactory() { // dataSource实现类变为PooledDataSource this.dataSource = new PooledDataSource(); } }
unPooledDataSourceFactory主要工作是对数据源进行参数配置，并提供获取数据源方法。分析PooledDataSourceFactory源码，只是继承unPooledDataSourceFactory，将DataSource实现类改变为PooledDataSource。
unPooledDataSource源码分析：基本的数据源实现都实现了DataSource接口，重写获取数据库连接的方法。unPooledDataSource从类名可知，不支持数据库连接的池化。也就是说，每来一个获取连接请求，就新建一个数据库连接。让我们看源码验证下。
public class UnpooledDataSource implements DataSource { private ClassLoader driverClassLoader; // 数据库驱动类加载器 private Properties driverProperties; // 有关数据库驱动的参数 private static Map<String, Driver> registeredDrivers = new ConcurrentHashMap<String, Driver>(); // 缓存已注册过的数据库驱动 private String driver; // 数据库驱动 private String url; // 数据库名 private String username; // 连接用户名 private String password; // 密码 private Boolean autoCommit; // 是否自动提交 private Integer defaultTransactionIsolationLevel; // 事物隔离级别 static { // 初始化 Enumeration<Driver> drivers = DriverManager.getDrivers(); // DriverManager中已存在的数据库驱动加载到数据库驱动缓存 while (drivers.hasMoreElements()) { Driver driver = drivers.nextElement(); registeredDrivers.put(driver.getClass().getName(), driver); } } ..... public Connection getConnection() throws SQLException { return doGetConnection(username, password); } // 获取数据库连接 private Connection doGetConnection(Properties properties) throws SQLException { initializeDriver(); // 初始化数据库驱动 Connection connection = DriverManager.getConnection(url, properties); // 此处每次获取连接，就新建一个数据库连接 configureConnection(connection); // 设置数据库是否自动提交，设置数据库事物隔离级别 return connection; } private synchronized void initializeDriver() throws SQLException { // 若此驱动还没初始化，则进行初始化 if (!registeredDrivers.containsKey(driver)) { Class<?> driverType; try { if (driverClassLoader != null) { driverType = Class.forName(driver, true, driverClassLoader); } else { driverType = Resources.classForName(driver); } // DriverManager requires the driver to be loaded via the system ClassLoader. // //www.kfu.com/~nsayer/Java/dyn-jdbc.html Driver driverInstance = (Driver)driverType.newInstance(); DriverManager.registerDriver(new DriverProxy(driverInstance)); registeredDrivers.put(driver, driverInstance); } catch (Exception e) { throw new SQLException("Error setting driver on UnpooledDataSource. Cause: " + e); } } } private void configureConnection(Connection conn) throws SQLException { if (autoCommit != null && autoCommit != conn.getAutoCommit()) { conn.setAutoCommit(autoCommit); } if (defaultTransactionIsolationLevel != null) { conn.setTransactionIsolation(defaultTransactionIsolationLevel); } } .... }
以上代码是UnPooledDataSource的源码分析，可见，UnPooledDataSource并没有采用池化的方法对数据库连接进行管理。每次获取连接，就新建一个数据库连接。我们知道数据库连接的建立是个非常耗时耗资源的过程，为了统一管理这些数据库连接，mybatis为我们引入了PooledDataSource类。
PooledDataSource源码分析：PooledDataSource是数据源的重点，源码比较复杂。PooledDataSource内部使用UnPooledDataSource类创建新的数据库连接。PooledDataSource并不直接管理java.sql.connection连接，而是管理java.sql.connection的一个代理类PooledConnection。除了管理数据库连接的建立，PooledDataSource内部还使用PoolState来管理数据源的状态（即空闲连接数，活跃连接数等）。综上，总结如下，PooledDataSource使用UnPooledDataSource类为数据源创建真实的数据库连接，使用PooledConnection为数据源管理数据库连接，使用PoolState来为数据源管理数据源当前状态。
PoolConnection是一个connection代理类，里面封装了真实的连接与代理连接，现在我们先来分析PoolConnection的源码。
class PooledConnection implements InvocationHandler { // 连接代理类 private static final String CLOSE = "close"; private static final Class<?>[] IFACES = new Class<?>[] { Connection.class }; private int hashCode = 0; private PooledDataSource dataSource; // 数据源 private Connection realConnection; // 被代理的真实连接 private Connection proxyConnection; // 代理连接 private long checkoutTimestamp; // 从连接池中取出连接的时间 private long createdTimestamp; // 连接建立的时间 private long lastUsedTimestamp; // 连接上次使用的时间 private int connectionTypeCode; // 用于标注该连接所在的连接池 private boolean valid; // 连接有效的标志
PooledConnection实现了InvocationHandler接口，则可见是一个代理对象。查看属性可知，内部有真实连接与代理连接，并附带连接的一些记录信息。查看该类的构造方法。
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); // 使用动态代理生成连接的代理类 } /* * Invalidates the connection */ // 将该链接置为无效 public void invalidate() { valid = false; }
查看构造方法可知，内部除了初始化一些属性外，还将连接的代理类也进行初始化了。那代理类究竟做了什么，查看重写的invoke方法源码。
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); return null; } else { try { if (!Object.class.equals(method.getDeclaringClass())) { // 若要执行的方法不是object方法，则检查连接的有效性 // 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."); } }
由源码可知，代理连接在执行方法时，会先检查此连接的有效性，然后执行真实的方法。分析完PoolConnection后，对PoolState进行源码解析。
public class PoolState { // 连接池状态信息 protected PooledDataSource dataSource; // 此状态信息关联的数据源 protected final List<PooledConnection> idleConnections = new ArrayList<PooledConnection>(); // 空闲连接列表 protected final List<PooledConnection> activeConnections = new ArrayList<PooledConnection>(); // 活跃连接列表 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; // 失效的连接数
PoolState是对DataSource的状态管理类，主要包括如累计连接超时时间，失效连接的获取等一些状态信息的管理。除了包括一些数据库连接的记录信息外，内部还维护了两个数据库连接的列表idleConnections，activeConnections.。分别用来存放空闲的数据库连接列表，活跃的数据库连接列表，针对此两个列表的操作，下文在分析PooledDataSource时会进行详细介绍。
对PoolConnection和PoolState分析结束后，具体分析PoolDataSource源码。
public class PooledDataSource implements DataSource { private static final Log log = LogFactory.getLog(PooledDataSource.class); private final PoolState state = new PoolState(this); // 维护数据源的状态 private final UnpooledDataSource dataSource; // 使用UnpooledDataSource来建立真正的连接 // OPTIONAL CONFIGURATION FIELDS protected int poolMaximumActiveConnections = 10; // 最大活跃的连接数 protected int poolMaximumIdleConnections = 5; // 最大空闲的连接数 protected int poolMaximumCheckoutTime = 20000; // 最大checkout时间（checkOutTime指的是从数据源中获取连接到归还连接的时间） protected int poolTimeToWait = 20000; // 最大等待时间 protected String poolPingQuery = "NO PING QUERY SET"; // 使用该语句来验证该连接是否有效 protected boolean poolPingEnabled = false; protected int poolPingConnectionsNotUsedFor = 0; private int expectedConnectionTypeCode; // hashcode
查看PoolDataSource基本属性，可知内部使用PoolState来维护数据源的状态信息，使用UnpooledDataSource来产真正的连接。并提供了一些如设置最大空闲，活跃连接数的配置信息。作为DataSource的实现，PooledDataSource不仅提供了如popConnection获取数据库连接的接口。还提供了forceCloseAll来关闭所有数据连接。pushConnection将使用结束的数据库连接放入数据源中。现在开始分析第一个方法popConnection，流程图如下，代码中都有详细注释，请耐看。
// 获取连接 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.size() > 0) { // 连接池中是否有空闲连接 // Pool has available connection conn = state.idleConnections.remove(0); // 从空闲连接列表中取一个空闲连接 if (log.isDebugEnabled()) { log.debug("Checked out connection " + conn.getRealHashCode() + " from pool."); } } else { // 连接池无空闲连接 // Pool does not have available connection if (state.activeConnections.size() < poolMaximumActiveConnections) { // 当前活跃连接数小于连接池的最大活跃连接数 // Can create new connection conn = new PooledConnection(dataSource.getConnection(), this); // 则使用unPooledDataSource新建一个连接,并封装成代理连接PooledConnection @SuppressWarnings("unused") //used in logging, if enabled Connection realConn = conn.getRealConnection(); // 获取真正的连接 if (log.isDebugEnabled()) { log.debug("Created connection " + conn.getRealHashCode() + "."); } } else { // 否则当前活跃连接数大于等于连接池的最大活跃连接数 // Cannot create new connection PooledConnection oldestActiveConnection = state.activeConnections.get(0); // 从活跃连接列表中取第一个活跃连接 long longestCheckoutTime = oldestActiveConnection.getCheckoutTime(); // 获取连接已经获取了多长时间 if (longestCheckoutTime > poolMaximumCheckoutTime) { // 检测该连接是否超时 // Can claim overdue connection 超时则进行统计 state.claimedOverdueConnectionCount++; state.accumulatedCheckoutTimeOfOverdueConnections += longestCheckoutTime; state.accumulatedCheckoutTime += longestCheckoutTime; state.activeConnections.remove(oldestActiveConnection); // 将此超时连接从活跃连接列表中移除 if (!oldestActiveConnection.getRealConnection().getAutoCommit()) { // 超时且关闭了自动提交，则进行回滚 oldestActiveConnection.getRealConnection().rollback(); } conn = new PooledConnection(oldestActiveConnection.getRealConnection(), this); // 新建一个代理连接 oldestActiveConnection.invalidate(); // 将超时的连接设置为失效 if (log.isDebugEnabled()) { log.debug("Claimed overdue connection " + conn.getRealHashCode() + "."); } } else { // 若没有一个连接超时，则必须等待 // Must wait try { if (!countedWait) { state.hadToWaitCount++; // 等待数加一 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) { 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++; conn = null; // 置为空，开始下一次循环 if (localBadConnectionCount > (poolMaximumIdleConnections + 3)) { 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; }
经分析，获取连接的过程为，先去查找空闲连接列表，若存在空闲列表，则直接从空闲列表中拿出数据库连接。若无空闲连接，则判断当前存活的数据库连接是否超过了指定的活跃连接数，若没有超过，则新建数据库连接。若超过了，则去拿活跃连接数的第一个连接判断是否连接超时（为什么拿第一个？因为是队列，队尾插入，对头获取，对头的连接没有超时，则后面的肯定没有超时）若发现第一个连接已经超过指定的数据库连接时间，则将此连接从活跃列表中移除，并标志为失效，然后自己新建一个数据库连接。若第一个连接没有过期，则代表现在数据源不能提供任何连接了，必须等待，直接wait，释放锁，等待线程唤醒。拿到了数据库连接后，需要检查该连接是否有效，若有效，则放入活跃连接列表中，并返回给用户。
当一个连接使用完毕后，需要放回到数据源中进行管理，现在分析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++; } } }
连接使用结束后并不是立马释放，而是检查当前空闲列表的连接数是否已超过指定空闲的连接数，若没有超过，则放入到空闲连接列表中。否则将该连接设为无效。并唤醒阻塞中的获取连接的线程。
当用户指定变更数据源配置信息时，如数据库地址，用户名，密码等，都需要对数据源进行重置，清空现存的数据库连接后修改配置信息。现查看清空数据源的方法forceCloseAll源码。此方法较简单，就不贴流程图了。
//关闭池中所有的活跃连接和空闲连接 public void forceCloseAll() { synchronized (state) { expectedConnectionTypeCode = assembleConnectionTypeCode(dataSource.getUrl(), dataSource.getUsername(), dataSource.getPassword()); for (int i = state.activeConnections.size(); i > 0; i--) { // 获取所有的活跃连接 try { PooledConnection conn = state.activeConnections.remove(i - 1); // 移除 conn.invalidate(); // 失效 Connection realConn = conn.getRealConnection(); if (!realConn.getAutoCommit()) { // 事务回滚 realConn.rollback(); } realConn.close(); } catch (Exception e) { // ignore } } for (int i = state.idleConnections.size(); i > 0; i--) { // 获取所有的空闲连接 try { PooledConnection conn = state.idleConnections.remove(i - 1); // 移除 conn.invalidate(); // 失效 Connection realConn = conn.getRealConnection(); if (!realConn.getAutoCommit()) { // 事务回滚 realConn.rollback(); } realConn.close(); } catch (Exception e) { // ignore } } } if (log.isDebugEnabled()) { log.debug("PooledDataSource forcefully closed/removed all connections."); } }
经分析，forceCloseAll对所有的空闲列表中，活跃列表中的数据库连接全部移除并置为不可用。池中恢复到初始化状态。
总结：本文对mybatis中的数据源部分进行了源码解析。在学习源码的过程中，加深了对很多设计模式的理解，体会到了大神们的编程习惯，不仅仅是源码本身，更多的是思想上的理解。在学习中也知道了不急于求成，一个一个的包去分析，然后再去整和业务流程。如你对此源码也感兴趣，可以评论下，我会把自己的mybatis中文注释源码包分享。但此注释都是自己手写，不能确保准确性，仅提供参考。任重而道远，源码之路希望自己能坚持下来。
