前言:我们使用mybatis时,关于数据源的配置多使用如c3p0,druid等第三方的数据源。其实mybatis内置了数据源的实现,提供了连接数据库,池的功能。在分析了缓存和日志包的源码后,接下来分析mybatis中的数据源实现。

类图:mybatis中关于数据源的源码包路径如下:

 myBatis源码解析-数据源篇(3)-LMLPHP

mybatis中提供了一个DataSourceFactory接口,提供了设置数据源配置信息,获取数据源方法。查看类图可知,有三个实现类分别提供了不同的数据源实现。JndiDataSourceFactory,PooledDataSourceFactory,unPooledDataSourceFactory。JndiDataSourceFactory实现较简单,此处源码略过。如下为各类的相互关系。

 myBatis源码解析-数据源篇(3)-LMLPHP

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.
        // http://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来为数据源管理数据源当前状态。

myBatis源码解析-数据源篇(3)-LMLPHP

 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,流程图如下,代码中都有详细注释,请耐看。

myBatis源码解析-数据源篇(3)-LMLPHP

 // 获取连接
  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源码。流程图和源码分析如下:

myBatis源码解析-数据源篇(3)-LMLPHP

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中文注释源码包分享。但此注释都是自己手写,不能确保准确性,仅提供参考。任重而道远,源码之路希望自己能坚持下来。

08-06 04:02