MyBatis源码分析(5)——内置DataSource实现

PooledConnection

此间最首假若因而Java Proxy代理来促成的。

// 实现代理接口
class PooledConnection implements InvocationHandler {

  private static final String CLOSE = "close";
  private static final Class<?>[] IFACES = new Class<?>[] { Connection.class };

  // 省略了部分代码...

  private PooledDataSource dataSource;
  private Connection realConnection;
  private Connection proxyConnection;

  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);
  }

  // 代理实现Connection接口的所有方法,只对CLOSE方法特别处理
  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())) {
          checkConnection();
        }
        // 其他方法则直接调用实际的连接来处理
        return method.invoke(realConnection, args);
      } catch (Throwable t) {
        throw ExceptionUtil.unwrapThrowable(t);
      }
    }
  }
}

PooledDataSource

pushConnection实现:
  // 省略了部分日志代码
  protected void pushConnection(PooledConnection conn) throws SQLException {
    synchronized (state) {
      // 从活动链表中移除当前连接
      state.activeConnections.remove(conn);
      if (conn.isValid()) {
        // 当前连接有效的话判断是否达到了最大空闲连接数,以及当前的连接是否变更过(即用户名,密码,Url等变更)
        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();
          // 通知等待获取连接的线程(不去判断是否真的有线程在等待)
          state.notifyAll();
        } else {
          state.accumulatedCheckoutTime += conn.getCheckoutTime();
          if (!conn.getRealConnection().getAutoCommit()) {
            conn.getRealConnection().rollback();
          }
          // 超出空闲连接限制,则直接释放当前连接
          conn.getRealConnection().close();
          // 将原有的代理连接设置为无效
          conn.invalidate();
        }
      } else {
        // 连接无效,则统计无效连接个数
        state.badConnectionCount++;
      }
    }
  }

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;

  public PoolState(PooledDataSource dataSource) {
    this.dataSource = dataSource;
  }

  // 省略了部分Getter方法...
}

配置

UNPOOLED数据源唯有5个属性需要配置:

driver:JDBC具体数据库驱动
url:JDBC连接
username:用户名
password:密码
defaultTransactionIsolationLevel:默认事务隔离级别

而外上述属性外,还是可以够配备驱动的总是音信,可是急需加前缀driver.,如:driver.encoding=UTF8,会将该配置作为参数传入driverManager.getConnection。

如下:

<dataSource type="UNPOOLED">
    <property name="driver" value="${jdbc.driver}" />
    <property name="url" value="${jdbc.url}" />
    <property name="username" value="${jdbc.username}" />
    <property name="password" value="${jdbc.password}" />
</dataSource>

PooledDataSource回收连接

透过利用代理,就足以在用户调用Close关闭数据库连接的时候,依据当下事态来控制放入空闲链表中仍然自由掉。

配置

除此之外UNPOOLED的配置外,还足以配备其余的有的属性,如下:

poolMaximumActiveConnections:最大活动连接数(默认为10)
poolMaximumIdleConnections:最大空闲连接数(默认为5)
poolMaximumCheckoutTime:最大可回收时间,即当达到最大活动链接数时,此时如果有程序获取连接,则检查最先使用的连接,看其是否超出了该时间,如果超出了该时间,则可以回收该连接。(默认20s)
poolTimeToWait:没有连接时,重尝试获取连接以及打印日志的时间间隔(默认20s)
poolPingQuery:检查连接正确的语句,默认为"NO PING QUERY SET",即没有,使用会导致抛异常
poolPingEnabled:是否开启ping检测,(默认:false)
poolPingConnectionsNotUsedFor:设置ping检测时间间隔,通常用于检测超时连接(默认为0,即当开启检测后每次从连接词中获取连接以及放回连接池都需要检测)

示范配置如下:

<dataSource type="POOLED">
    <property name="driver" value="${jdbc.driver}" />
    <property name="url" value="${jdbc.url}" />
    <property name="username" value="${jdbc.username}" />
    <property name="password" value="${jdbc.password}" />
    <property name="poolMaximumActiveConnections" value="20" />
    <property name="poolMaximumIdleConnections" value="10" />
    <property name="poolMaximumCheckoutTime" value="15" />
    <property name="poolTimeToWait" value="10" />
    <property name="poolPingQuery" value="select 1 from dual" />
    <property name="poolPingEnabled" value="true" />
    <property name="poolPingConnectionsNotUsedFor" value="0" />
</dataSource>

当日志开启DEBUG输出,使用Mapper操作时,能够看来:

2016-07-31 21:30:45 [DEBUG]-[Thread: main]-[org.apache.ibatis.datasource.pooled.PooledDataSource.popConnection()]: 
Created connection 395084181.

2016-07-31 21:30:45 [DEBUG]-[Thread: main]-[org.apache.ibatis.datasource.pooled.PooledDataSource.pingConnection()]: 
Testing connection 395084181 ...

2016-07-31 21:30:45 [DEBUG]-[Thread: main]-[org.apache.ibatis.datasource.pooled.PooledDataSource.pingConnection()]: 
Connection 395084181 is GOOD!

实现

UnpooledDataSource,内部通过调用DriverManager来实现,DriverManager是用于管理低层驱动以及开创连接的。

public class UnpooledDataSource implements DataSource {

  private ClassLoader driverClassLoader;
  // 驱动连接属性
  private Properties driverProperties;
  // 所有已注册的驱动,仅仅用于识别驱动在DriverManager中是否已经被加载进来了
  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 {
    // 静态代码块,当类加载的时候,就从DriverManager中获取所有的驱动信息,放到当前维护的Map中
    Enumeration<Driver> drivers = DriverManager.getDrivers();
    while (drivers.hasMoreElements()) {
      Driver driver = drivers.nextElement();
      registeredDrivers.put(driver.getClass().getName(), driver);
    }
  }

  // 省略了部分代码... 

  public Connection getConnection() throws SQLException {
    // 获取数据库连接
    return doGetConnection(username, password);
  }

  public Connection getConnection(String username, String password) throws SQLException {
    return doGetConnection(username, password);
  }

  private Connection doGetConnection(String username, String password) throws SQLException {
    // 这里通过url加Properties来获取连接,是因为可以在配置文件中配置数据库连接的信息,比如编码之类的
    Properties props = new Properties();
    if (driverProperties != null) {
      props.putAll(driverProperties);
    }
    if (username != null) {
      props.setProperty("user", username);
    }
    if (password != null) {
      props.setProperty("password", password);
    }
    return doGetConnection(props);
  }

  private Connection doGetConnection(Properties properties) throws SQLException {
    // 初始化驱动信息
    initializeDriver();
    // 从DriverManager中获取数据库连接
    Connection connection = DriverManager.getConnection(url, properties);
    // 配置连接信息,自动提交以及事务隔离级别
    configureConnection(connection);
    return connection;
  }

  private synchronized void initializeDriver() throws SQLException {
    // 如果没有包含在已注册Map中,则需要将该驱动加载进来
    if (!registeredDrivers.containsKey(driver)) {
      Class<?> driverType;
      try {
        // 加载数据库连接驱动
        if (driverClassLoader != null) {
          driverType = Class.forName(driver, true, driverClassLoader);
        } else {
          // Resources为MyBatis内置的资源工具类,该方法依次尝试从多个ClassLoader中获取Class类,顺序为:配置的classLoader,默认的defaultClassLoader,当前线程的getContextClassLoader,当前类的getClass().getClassLoader(),系统的systemClassLoader
          driverType = Resources.classForName(driver);
        }
        // 创建驱动实例
        Driver driverInstance = (Driver)driverType.newInstance();
        // 注册到DriverManager中,用于创建数据库连接
        DriverManager.registerDriver(new DriverProxy(driverInstance));
        // 放到已注册Map中
        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);
    }
  }
}

参考

  1. MyBatis官方文档

@(MyBatis)[DataSource]

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) {
          // 有空闲连接,直接获取
          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);
            Connection realConn = conn.getRealConnection();
            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++;
              // 统计过期回收时间增加
              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();
            }
            // 根据URL,用户名以及密码计算出一个Hash,用于标识此次连接
            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 + 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.");
            }
          }
        }
      }

    }

    // 从上面的循环退出,如果为null,则一定出现异常情况了
    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;
  }

MyBatis源码分析(5)——内置DataSource实现

MyBatis内置了六个DataSource的实现:UnpooledDataSource,该数据源对于每一遍得到请求都简单的打开和倒闭连接。PooledDataSource,该数据源在Unpooled的功底上构建了连接池。

MyBatis DataSource集成

// 后续看完Spring-MyBatis再补充

流程图如下:

图片 1

流程图如下:

图片 2

实现

连接池的贯彻,主题的沉思在于,将连接缓存起来,即可以通过多少个链表(/队列)来贯彻,一个用以维持活动连续,另一个保持空闲连接。还有此外一些就是关闭连接后回来给连接池或者释放掉,这里可以经过代理来兑现,当客户端调用close时,并不是直接关闭连接,而是将其缓存起来,放到空闲链表中。这里运用代理还有个优点,每一趟释放的时候,重新创立新的代理连接来封装,并且将原来的代办设置为无效,可以使得程序即使拥有原有的代办,也不会影响到回收的总是。
在MyBatis中,首如果因而PoolState来珍爱连接情状,以及通过PooledConnection代办来促成归还连接操作。


连接池调优

// 经验不足,后续补充

PooledDataSource拿到连接

连接池紧如果由此popConnection来兑现连续的创造以及分配的,过程不复杂,当有空余连接时则一贯利用,否则再遵照是否达到了安装的峰值,来支配是否需要创立新的连续等。

UnpooledDataSource

网站地图xml地图