MyBatis 数据源原理 - litter-fish/ReadSource GitHub Wiki
配置连接池
MyBatis 支持三种数据源配置，分别为 UNPOOLED、POOLED 和 JNDI
- UNPOOLED - 每次请求时都会重新打开和关闭新的连接。
- POOLED - 利用“池”的概念将 JDBC 连接对象组织起来，避免创建新的连接实例时所必需的初始化和认证时间。
- JNDI - 能在如 EJB 或应用服务器这类容器中使用，容器可以集中或在外部配置数据源，然后放置一个 JNDI 上下文的引用。
<environments default="development">
<environment id="development">
<!-- 事务管理器的配置 -->
<transactionManager type="JDBC">
<property name="..." value="..."/>
</transactionManager>
<!-- 数据源的配置 -->
<dataSource type="POOLED|UNPOOLED">
<property name="driver" value="${driver}"/>
<property name="url" value="${url}"/>
<property name="username" value="${username}"/>
<property name="password" value="${password}"/>
</dataSource>
</environment></environments>
数据源类图
数据源初始化过程
开始解析environments节点
// XMLConfigBuilder
private void environmentsElement(XNode context) throws Exception {
if (context != null) {
if (environment == null) {
// 默认使用的环境 ID
environment = context.getStringAttribute("default");
}
for (XNode child : context.getChildren()) {
// 每个 environment 元素定义的环境 ID
String id = child.getStringAttribute("id");
if (isSpecifiedEnvironment(id)) {
// 获取事务管理器的配置
TransactionFactory txFactory = transactionManagerElement(child.evalNode("transactionManager"));
// 获取数据源工厂类实例
DataSourceFactory dsFactory = dataSourceElement(child.evalNode("dataSource"));
DataSource dataSource = dsFactory.getDataSource();
Environment.Builder environmentBuilder = new Environment.Builder(id)
.transactionFactory(txFactory)
.dataSource(dataSource);
configuration.setEnvironment(environmentBuilder.build());
}
}
}
}
初始化数据源工厂类实例
// XMLConfigBuilder
private DataSourceFactory dataSourceElement(XNode context) throws Exception {
if (context != null) {
// 获取数据源类型：类型，UNPOOLED|POOLED|JNDI
// UNPOOLED– 这个数据源的实现只是每次被请求时打开和关闭连接
// POOLED– 这种数据源的实现利用“池”的概念将 JDBC 连接对象组织起来，避免了创建新的连接实例时所必需的初始化和认证时间。
// JNDI – 这个数据源的实现是为了能在如 EJB 或应用服务器这类容器中使用，容器可以集中或在外部配置数据源，然后放置一个 JNDI 上下文的引用。
String type = context.getStringAttribute("type");
// 获取特定数据类型的属性
Properties props = context.getChildrenAsProperties();
// 获取别名，通过反射创建实例
DataSourceFactory factory = (DataSourceFactory) resolveClass(type).newInstance();
// 赋值数据库连接池工厂属性
factory.setProperties(props);
return factory;
}
throw new BuilderException("Environment declaration requires a DataSourceFactory.");
}
根据数据源类型创建数据源工厂类
public class PooledDataSourceFactory extends UnpooledDataSourceFactory {
public PooledDataSourceFactory() {
// 创建 PooledDataSource
this.dataSource = new PooledDataSource();
}
}
public class PooledDataSource implements DataSource {
public PooledDataSource() {
dataSource = new UnpooledDataSource();
}
// 省略其他代码
}
设置数据源工厂属性值
// org/apache/ibatis/datasource/unpooled/UnpooledDataSourceFactory.java
public void setProperties(Properties properties) {
Properties driverProperties = new Properties();
// 为 dataSource 创建元信息对象
MetaObject metaDataSource = SystemMetaObject.forObject(dataSource);
// 遍历 properties 键列表，properties 由配置文件解析器传入
for (Object key : properties.keySet()) {
String propertyName = (String) key;
// 检测 propertyName 是否以 "driver." 开头
if (propertyName.startsWith(DRIVER_PROPERTY_PREFIX)) {
String value = properties.getProperty(propertyName);
// 存储配置信息到 driverProperties 中
driverProperties.setProperty(propertyName.substring(DRIVER_PROPERTY_PREFIX_LENGTH), value);
} else if (metaDataSource.hasSetter(propertyName)) {
String value = (String) properties.get(propertyName);
// 按需转换 value 类型
Object convertedValue = convertValue(metaDataSource, propertyName, value);
// 设置转换后的值到 UnpooledDataSourceFactory 指定属性中
metaDataSource.setValue(propertyName, convertedValue);
} else {
throw new DataSourceException("Unknown DataSource property: " + propertyName);
}
}
if (driverProperties.size() > 0) {
// 设置 driverProperties 到 UnpooledDataSourceFactory 的 driverProperties 属性中
metaDataSource.setValue("driverProperties", driverProperties);
}
}
按需转换 value 类型
// org/apache/ibatis/datasource/unpooled/UnpooledDataSourceFactory.java
private Object convertValue(MetaObject metaDataSource, String propertyName, String value) {
Object convertedValue = value;
// 获取属性对应的 setter 方法的参数类型
Class<?> targetType = metaDataSource.getSetterType(propertyName);
// 按照 setter 方法的参数类型进行类型转换
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;
}
获取数据库连接
执行crud操作时，在创建 Statement 步骤时会获取JDBC连接
// org/apache/ibatis/executor/SimpleExecutor.java
private Statement prepareStatement(StatementHandler handler, Log statementLog) throws SQLException {
Statement stmt;
// 获取JDBC连接
Connection connection = getConnection(statementLog);
// 调用语句处理器的prepare方法
stmt = handler.prepare(connection, transaction.getTimeout());
// 为 Statement 设置 IN 参数
handler.parameterize(stmt);
return stmt;
}
获取JDBC连接
protected Connection getConnection(Log statementLog) throws SQLException {
// 通过jdbc获取数据库连接
Connection connection = transaction.getConnection();
if (statementLog.isDebugEnabled()) {
return ConnectionLogger.newInstance(connection, statementLog, queryStack);
} else {
return connection;
}
}
委托事务管理器获取连接
// org/apache/ibatis/transaction/jdbc/JdbcTransaction.java
public Connection getConnection() throws SQLException {
if (connection == null) {
openConnection();
}
return connection;
}
打开连接，设置自动提交标识
// org/apache/ibatis/transaction/jdbc/JdbcTransaction.java
protected void openConnection() throws SQLException {
if (log.isDebugEnabled()) {
log.debug("Opening JDBC Connection");
}
// 获取连接
connection = dataSource.getConnection();
if (level != null) {
// 设置隔离级别
connection.setTransactionIsolation(level.getLevel());
}
// 设置提交模式
setDesiredAutoCommit(autoCommmit);
}
非池化数据源获取连接
获取非池化连接的时序图：
该种数据源不具有池化特性。该种数据源每次会返回一个新的数据库连接，而非复用旧的连接
// org/apache/ibatis/datasource/unpooled/UnpooledDataSource.java
public Connection getConnection() throws SQLException {
// 通过用户名和密码获取连接
return doGetConnection(username, password);
}
通过用户名和密码获取连接
// org/apache/ibatis/datasource/unpooled/UnpooledDataSource.java
private Connection doGetConnection(String username, String password) throws SQLException {
Properties props = new Properties();
if (driverProperties != null) {
props.putAll(driverProperties);
}
if (username != null) {
// 存储 user 配置
props.setProperty("user", username);
}
if (password != null) {
// 存储 password 配置
props.setProperty("password", password);
}
// 调用重载方法
return doGetConnection(props);
}
设置用户名和密码到Properties对象后调用重载方法获取链接
// org/apache/ibatis/datasource/unpooled/UnpooledDataSource.java
private Connection doGetConnection(Properties properties) throws SQLException {
// 初始化数据库驱动
initializeDriver();
// 获取连接
Connection connection = DriverManager.getConnection(url, properties);
// 配置连接，包括自动提交以及事务等级
configureConnection(connection);
return connection;
}
初始化数据库驱动
// org/apache/ibatis/datasource/unpooled/UnpooledDataSource.java
private synchronized void initializeDriver() throws SQLException {
// 检测缓存中是否包含了与 driver 对应的驱动实例
if (!registeredDrivers.containsKey(driver)) {
Class<?> driverType;
try {
// 加载驱动类型
if (driverClassLoader != null) {
// 使用 driverClassLoader 加载驱动
driverType = Class.forName(driver, true, driverClassLoader);
} else {
// 通过其他 ClassLoader 加载驱动
driverType = Resources.classForName(driver);
}
// DriverManager requires the driver to be loaded via the system ClassLoader.
// http://www.kfu.com/~nsayer/Java/dyn-jdbc.html
// 通过其他 ClassLoader 加载驱动
Driver driverInstance = (Driver)driverType.newInstance();
/*
* 注册驱动，注意这里是将 Driver 代理类 DriverProxy 对象注册到 DriverManager 中的，
* 而非 Driver 对象本身。DriverProxy 中并没什么特别的逻辑，就不分析。
*/
DriverManager.registerDriver(new DriverProxy(driverInstance));
// 缓存驱动类名和实例
registeredDrivers.put(driver, driverInstance);
} catch (Exception e) {
throw new SQLException("Error setting driver on UnpooledDataSource. Cause: " + e);
}
}
}
配置连接，包括自动提交以及事务等级
// org/apache/ibatis/datasource/unpooled/UnpooledDataSource.java
private void configureConnection(Connection conn) throws SQLException {
if (autoCommit != null && autoCommit != conn.getAutoCommit()) {
// 设置自动提交
conn.setAutoCommit(autoCommit);
}
if (defaultTransactionIsolationLevel != null) {
// 设置事务隔离级别
conn.setTransactionIsolation(defaultTransactionIsolationLevel);
}
}
池化数据源 PooledDataSource
辅助类介绍
- PoolState 用于记录连接池运行时的状态，比如连接获取次数，无效连接数量等。同时 PoolState 内部定义了两个 PooledConnection 集合，用于存储空闲连接和活跃连接
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;
public PoolState(PooledDataSource dataSource) {
this.dataSource = dataSource;
}
// 省略部分代码
}
- PooledConnection 内部定义了一个 Connection 类型的变量，用于指向真实的数据库连接。以及一个 Connection 的代理类，用于对部分方法调用进行拦截。
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;
// connectionTypeCode = (url + username + password).hashCode()
private int connectionTypeCode;
// 表示连接是否有效
private boolean valid;
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;
// 创建 Connection 的代理类对象
this.proxyConnection = (Connection) Proxy.newProxyInstance(Connection.class.getClassLoader(), IFACES, this);
}
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {...}
// 省略部分代码
}
池化数据连接的获取时序图：
时序图中只画出创建池连接的过程。
获取连接
// org/apache/ibatis/datasource/pooled/PooledDataSource.java
public Connection getConnection() throws SQLException {
// 返回 Connection 的代理对象
return popConnection(dataSource.getUsername(), dataSource.getPassword()).getProxyConnection();
}
先获取连接
// org/apache/ibatis/datasource/pooled/PooledDataSource.java
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) {
// 检测空闲连接集合（idleConnections）是否为空
if (!state.idleConnections.isEmpty()) {
// idleConnections 不为空，表示有空闲连接可以使用
conn = state.idleConnections.remove(0);
if (log.isDebugEnabled()) {
log.debug("Checked out connection " + conn.getRealHashCode() + " from pool.");
}
} else {
/*
* 暂无空闲连接可用，但如果活跃连接数还未超出限制
*（poolMaximumActiveConnections），则可创建新的连接
*/
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) {
// Can claim overdue connection
// 累加超时相关的统计字段
state.claimedOverdueConnectionCount++;
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");
}
}
/*
* 创建一个新的 PooledConnection，注意，
* 此处复用 oldestActiveConnection 的 realConnection 变量
*/
conn = new PooledConnection(oldestActiveConnection.getRealConnection(), this);
/*
* 复用 oldestActiveConnection 的一些信息，注意 PooledConnection 中的
* createdTimestamp 用于记录 Connection 的创建时间，而非 PooledConnection
* 的创建时间。所以这里要复用原连接的时间信息。
*/
conn.setCreatedTimestamp(oldestActiveConnection.getCreatedTimestamp());
conn.setLastUsedTimestamp(oldestActiveConnection.getLastUsedTimestamp());
// 设置旧的连接为无效状态
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) {
// ping to server and check the connection is valid or not
/*
* 检测连接是否有效，isValid 方法除了会检测 valid 是否为 true，
* 还会通过 PooledConnection 的 pingConnection 方法执行 SQL 语句，
* 检测连接是否可用。pingConnection 方法的逻辑不复杂，大家可以自行分析。
* 另外，官方文档在介绍 POOLED 类型数据源时，也介绍了连接有效性检测方面的
* 属性，有三个：poolPingQuery，poolPingEnabled 和
* poolPingConnectionsNotUsedFor。关于这三个属性，大家可以查阅官方文档
*/
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++;
localBadConnectionCount++;
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;
}
以上逻辑的伪代码实现
if (连接池中有空闲连接) {
1. 将连接从空闲连接集合中移除并返回
} else {
if (活跃连接数未超出限制) {
1. 创建新连接
} else {
1. 从活跃连接集合中取出第一个元素
2. 获取连接运行时长
if (连接超时) {
1. 将连接从活跃集合中移除
2. 复用原连接的成员变量，并创建新的 PooledConnection 对象
} else {
1. 线程进入等待状态
2. 线程被唤醒后，重新执行以上逻辑
}
}
}
popConnection 实现流程图
创建一个新的 PooledConnection
// org/apache/ibatis/datasource/pooled/PooledConnection.java
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;
// 创建 Connection 的代理类对象
this.proxyConnection = (Connection) Proxy.newProxyInstance(Connection.class.getClassLoader(), IFACES, this);
}
检测连接是否有效
// org/apache/ibatis/datasource/pooled/PooledConnection.java
public boolean isValid() {
return valid && realConnection != null && dataSource.pingConnection(this);
}
通过PING检测连接是否有效
// org/apache/ibatis/datasource/pooled/PooledDataSource.java
protected boolean pingConnection(PooledConnection conn) {
boolean result = true;
try {
result = !conn.getRealConnection().isClosed();
} catch (SQLException e) {
if (log.isDebugEnabled()) {
log.debug("Connection " + conn.getRealHashCode() + " is BAD: " + e.getMessage());
}
result = false;
}
if (result) {
if (poolPingEnabled) {
// ping检测时间间隔 >= 0 && 最后一次使用时间 > ping检测时间间隔
if (poolPingConnectionsNotUsedFor >= 0 && conn.getTimeElapsedSinceLastUse() > poolPingConnectionsNotUsedFor) {
try {
if (log.isDebugEnabled()) {
log.debug("Testing connection " + conn.getRealHashCode() + " ...");
}
Connection realConn = conn.getRealConnection();
Statement statement = realConn.createStatement();
// 检查连接正确的语句，默认为"NO PING QUERY SET"，即没有，使用会导致抛异常
ResultSet rs = statement.executeQuery(poolPingQuery);
rs.close();
statement.close();
if (!realConn.getAutoCommit()) {
realConn.rollback();
}
result = true;
if (log.isDebugEnabled()) {
log.debug("Connection " + conn.getRealHashCode() + " is GOOD!");
}
} catch (Exception e) {
log.warn("Execution of ping query '" + poolPingQuery + "' failed: " + e.getMessage());
try {
conn.getRealConnection().close();
} catch (Exception e2) {
//ignore
}
result = false;
if (log.isDebugEnabled()) {
log.debug("Connection " + conn.getRealHashCode() + " is BAD: " + e.getMessage());
}
}
}
}
}
return result;
}
回收连接，获取连接时，会通过JDK的动态代理创建一个连接的代理对象，当调用连接关闭时会触发回调invoke方法
// org/apache/ibatis/datasource/pooled/PooledConnection.java
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
String methodName = method.getName();
// 检测 close 方法是否被调用，若被调用则拦截之
if (CLOSE.hashCode() == methodName.hashCode() && CLOSE.equals(methodName)) {
// 将回收连接中，而不是直接将连接关闭
dataSource.pushConnection(this);
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);
}
}
}
回收连接
// org/apache/ibatis/datasource/pooled/PooledDataSource.java
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
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++;
}
}
}
回收连接流程图
