手写mybatis之数据源池化技术实现

前言

在上一章节我们解析了 XML 中数据源配置信息，并使用 Druid 创建数据源完成数据库的操作。但其实在 Mybatis 中是有自己的数据源实现的，包括无池化的 UnpooledDataSource 实现方式和有池化的 PooledDataSource 实现方式。

你可以把池化技术理解为享元模式的具体实现方案，通常我们对一些需要较高创建成本且高频使用的资源，需要进行缓存或者也称预热处理。并把这些资源存放到一个预热池子中，需要用的时候从池子中获取，使用完毕再进行使用。通过池化可以非常有效的控制资源的使用成本，包括；资源数量、空闲时长、获取方式等进行统一控制和管理。

在这里插入图片描述
此外由于控制了连接池中连接的数量，所以当外部从连接池获取链接时，如果链接已满则会进行循环等待。这也是大家日常使用DB连接池，如果一个SQL操作引起了慢查询，则会导致整个服务进入瘫痪的阶段，各个和数据库相关的接口调用，都不能获得到链接，接口查询TP99陡然增高，系统开始大量报警。那连接池可以配置的很大吗，也不可以，因为连接池要和数据库所分配的连接池对应上，避免应用配置连接池超过数据库所提供的连接池数量，否则会出现夯住不能分配链接的问题，导致数据库拖垮从而引起连锁反应。
无池化链接实现
对于数据库连接池的实现，不一定非得提供池化技术，对于某些场景可以只使用无池化的连接池。那么在实现的过程中，可以把无池化的实现和池化实现拆分解耦，在需要的时候只需要配置对应的数据源即可。

public class UnpooledDataSource implements DataSource {

    private ClassLoader driverClassLoader;
    // 驱动配置，也可以扩展属性信息 driver.encoding=UTF8
    private Properties driverProperties;
    // 驱动注册器
    private static Map<String, Driver> registeredDrivers = new ConcurrentHashMap<>();
    // 驱动
    private String driver;
    // DB 链接地址
    private String url;
    // 账号
    private String username;
    // 密码
    private String password;
    // 是否自动提交
    private Boolean autoCommit;
    // 事务级别
    private Integer defaultTransactionIsolationLevel;

    static {
        Enumeration<Driver> drivers = DriverManager.getDrivers();
        while (drivers.hasMoreElements()) {
            Driver driver = drivers.nextElement();
            registeredDrivers.put(driver.getClass().getName(), driver);
        }
    }

    private Connection doGetConnection(Properties properties) throws SQLException {
        initializerDriver();
        Connection connection = DriverManager.getConnection(url, properties);
        if (autoCommit != null && autoCommit != connection.getAutoCommit()) {
            connection.setAutoCommit(autoCommit);
        }
        if (defaultTransactionIsolationLevel != null) {
            connection.setTransactionIsolation(defaultTransactionIsolationLevel);
        }
        return connection;
    }

    /**
     * 初始化驱动
     */
    private synchronized void initializerDriver() throws SQLException {
        if (!registeredDrivers.containsKey(driver)) {
            try {
                Class<?> driverType = Class.forName(driver, true, driverClassLoader);
                // https://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);
            }
        }
    }
	
}

1：无池化的数据源链接实现比较简单，核心在于 initializerDriver 初始化驱动中使用了 Class.forName 和 newInstance 的方式创建了数据源链接操作。
2：在创建完成连接以后，把链接存放到驱动注册器中，方便后续使用中可以直接获取链接，避免重复创建所带来的资源消耗。
有池化链接实现
由于我们需要对连接进行池化处理，所以当链接调用一些 CLOSE 方法的时候，也需要把链接从池中关闭和恢复可用，允许其他用户获取到链接。那么这里就需要对连接类进行代理包装，处理 CLOSE 方法。

public class PooledConnection implements InvocationHandler {

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        String methodName = method.getName();
        // 如果是调用 CLOSE 关闭链接方法，则将链接加入连接池中，并返回null
        if (CLOSE.hashCode() == methodName.hashCode() && CLOSE.equals(methodName)) {
            dataSource.pushConnection(this);
            return null;
        } else {
            if (!Object.class.equals(method.getDeclaringClass())) {
                // 除了toString()方法，其他方法调用之前要检查connection是否还是合法的,不合法要抛出SQLException
                checkConnection();
            }
            // 其他方法交给connection去调用
            return method.invoke(realConnection, args);
        }
    }
    
}

1：在 invoke 方法中处理对 CLOSE 方法控制以外，排除 toString 等Object 的方法后，则是其他真正需要被 DB 链接处理的方法了。
2：那么这里有一个对于 CLOSE 方法的数据源回收操作 dataSource.pushConnection(this); 有一个具体的实现方法，在池化实现类 PooledDataSource 中进行处理。
pushConnection 回收链接

protected void pushConnection(PooledConnection connection) throws SQLException {
    synchronized (state) {
        state.activeConnections.remove(connection);
        // 判断链接是否有效
        if (connection.isValid()) {
            // 如果空闲链接小于设定数量，也就是太少时
            if (state.idleConnections.size() < poolMaximumIdleConnections && connection.getConnectionTypeCode() == expectedConnectionTypeCode) {
                state.accumulatedCheckoutTime += connection.getCheckoutTime();
                if (!connection.getRealConnection().getAutoCommit()) {
                    connection.getRealConnection().rollback();
                }
                // 实例化一个新的DB连接，加入到idle列表
                PooledConnection newConnection = new PooledConnection(connection.getRealConnection(), this);
                state.idleConnections.add(newConnection);
                newConnection.setCreatedTimestamp(connection.getCreatedTimestamp());
                newConnection.setLastUsedTimestamp(connection.getLastUsedTimestamp());
                connection.invalidate();
                logger.info("Returned connection " + newConnection.getRealHashCode() + " to pool.");
                // 通知其他线程可以来抢DB连接了
                state.notifyAll();
            }
            // 否则，空闲链接还比较充足
            else {
                state.accumulatedCheckoutTime += connection.getCheckoutTime();
                if (!connection.getRealConnection().getAutoCommit()) {
                    connection.getRealConnection().rollback();
                }
                // 将connection关闭
                connection.getRealConnection().close();
                logger.info("Closed connection " + connection.getRealHashCode() + ".");
                connection.invalidate();
            }
        } else {
            logger.info("A bad connection (" + connection.getRealHashCode() + ") attempted to return to the pool, discarding connection.");
            state.badConnectionCount++;
        }
    }
}

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.isEmpty()) {
                conn = state.idleConnections.remove(0);
                logger.info("Checked out connection " + conn.getRealHashCode() + " from pool.");
            }
            // 如果无空闲链接：创建新的链接
            else {
                // 活跃连接数不足
                if (state.activeConnections.size() < poolMaximumActiveConnections) {
                    conn = new PooledConnection(dataSource.getConnection(), this);
                    logger.info("Created connection " + conn.getRealHashCode() + ".");
                }
                // 活跃连接数已满
                else {
                    // 取得活跃链接列表的第一个，也就是最老的一个连接
                    PooledConnection oldestActiveConnection = state.activeConnections.get(0);
                    long longestCheckoutTime = oldestActiveConnection.getCheckoutTime();
                    // 如果checkout时间过长，则这个链接标记为过期
                    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();
                        logger.info("Claimed overdue connection " + conn.getRealHashCode() + ".");
                    }
                    // 如果checkout超时时间不够长，则等待
                    else {
                        try {
                            if (!countedWait) {
                                state.hadToWaitCount++;
                                countedWait = true;
                            }
                            logger.info("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));
                    // 记录checkout时间
                    conn.setCheckoutTimestamp(System.currentTimeMillis());
                    conn.setLastUsedTimestamp(System.currentTimeMillis());
                    state.activeConnections.add(conn);
                    state.requestCount++;
                    state.accumulatedRequestTime += System.currentTimeMillis() - t;
                } else {
                    logger.info("A bad connection (" + conn.getRealHashCode() + ") was returned from the pool, getting another connection.");
                    // 如果没拿到，统计信息：失败链接 +1
                    state.badConnectionCount++;
                    localBadConnectionCount++;
                    conn = null;
                    // 失败次数较多，抛异常
                    if (localBadConnectionCount > (poolMaximumIdleConnections + 3)) {
                        logger.debug("PooledDataSource: Could not get a good connection to the database.");
                        throw new SQLException("PooledDataSource: Could not get a good connection to the database.");
                    }
                }
            }
        }
    }

    return conn;
}

数据源工厂
数据源工厂包括两部分，分别是无池化和有池化，有池化的工厂继承无池化工厂，因为在 Mybatis 源码的实现类中，这样就可以减少对 Properties 统一包装的反射方式的属性处理。由于我们暂时没有对这块逻辑进行开发，只是简单的获取属性传参，所以还不能体现出这样的继承有多便捷，读者可以参考源码进行理解。源码类：UnpooledDataSourceFactory。
无池化工厂

public class UnpooledDataSourceFactory implements DataSourceFactory {

    protected Properties props;

    @Override
    public void setProperties(Properties props) {
        this.props = props;
    }

    @Override
    public DataSource getDataSource() {
        UnpooledDataSource unpooledDataSource = new UnpooledDataSource();
        unpooledDataSource.setDriver(props.getProperty("driver"));
        unpooledDataSource.setUrl(props.getProperty("url"));
        unpooledDataSource.setUsername(props.getProperty("username"));
        unpooledDataSource.setPassword(props.getProperty("password"));
        return unpooledDataSource;
    }

}

有池化工厂

public class PooledDataSourceFactory extends UnpooledDataSourceFactory {

    @Override
    public DataSource getDataSource() {
        PooledDataSource pooledDataSource = new PooledDataSource();
        pooledDataSource.setDriver(props.getProperty("driver"));
        pooledDataSource.setUrl(props.getProperty("url"));
        pooledDataSource.setUsername(props.getProperty("username"));
        pooledDataSource.setPassword(props.getProperty("password"));
        return pooledDataSource;
    }

}

测试
配置数据源

<environments default="development">
    <environment id="development">
        <transactionManager type="JDBC"/>
        <dataSource type="DRUID">
            <property name="driver" value="com.mysql.jdbc.Driver"/>
            <property name="url" value="jdbc:mysql://127.0.0.1:3306/mybatis?useUnicode=true"/>
            <property name="username" value="root"/>
            <property name="password" value="123456"/>
        </dataSource>
    </environment>
</environments>

配置Mapper

<select id="queryUserInfoById" parameterType="java.lang.Long" resultType="cn.bugstack.mybatis.test.po.User">
    SELECT id, userId, userName, userHead
    FROM user
    where id = #{id}
</select>

单元测试

@Test
public void test01() throws IOException {
    // 1. 从SqlSessionFactory中获取SqlSession
    SqlSessionFactory sqlSessionFactory = new SqlSessionFactoryBuilder().build(Resources.getResourceAsReader("mybatis-config-datasource.xml"));
    SqlSession sqlSession = sqlSessionFactory.openSession();
    
    // 2. 获取映射器对象
    IUserDao userDao = sqlSession.getMapper(IUserDao.class);
    
    // 3. 测试验证
    for (int i = 0; i < 50; i++) {
        User user = userDao.queryUserInfoById(1L);
        logger.info("测试结果：{}", JSON.toJSONString(user));
    }
}

无池化测试

<dataSource type="UNPOOLED"></dataSource>

有池化测试

<dataSource type="POOLED"></dataSource>

总结
1：我们完成了 Mybatis 数据源池化的设计和实现，也能通过这样的分析、实现、验证的过程让大家更好的理解我们平常使用的连接池所遇到的一些真实问题都是怎么发生的，做到知其然知其所以然。
另外关于连接池的实现重点可以随着调试验证的过程中进行学习，包括：synchronized 加锁、创建连接、活跃数量控制、休眠等待时长，抛异常逻辑等，这些都与我们日常使用连接池时的配置息息相关。
2：这一章节的内容可以算作是 Mybatis 核心功能实现过程上的重要分支，虽然可以使用 Druid 替代数据源的处理，但只有动手自己实现一遍数据源连接池才能更好的理解池化技术的落地方案，也能为以后做此类功能时，有一个可落地的具体方案。

好了到这里就结束了手写mybatis之数据源池化技术实现的学习，大家一定要跟着动手操作起来。需要源码的可si我获取；