Mybatis数据源类型和连接池实现原理源码分析

╰+攻爆jí腚メ 2023-06-07 12:11 11阅读 0赞

> Mybatis中的dataSource使用标准的 JDBC 数据源接口来配置 JDBC 连接对象的资源，有三种内建的数据源类型：
> 
>  *  UNPOOLED：这个数据源的实现只是每次被请求时打开和关闭连接。虽然有点慢，但对于在数据库连接可用性方面没有太高要求的简单应用程序来说，是一个很好的选择。 不同的数据库在性能方面的表现也是不一样的，对于某些数据库来说，使用连接池并不重要，这个配置就很适合这种情形。
>  *  POOLED：这个数据源的实现利用“池”的概念将 JDBC 连接对象组织起来，避免了创建新的连接实例时所必需的初始化和认证时间。 这是一种使得并发 Web 应用快速响应请求的流行处理方式。
>  *  JNDI：这个数据源的实现是为了能在如 EJB 或应用服务器这类容器中使用，容器可以集中或在外部配置数据源，然后放置一个 JNDI 上下文的引用。如，我们常在wildfly的standalone.xml 文件中集中配置 JNDI 数据源。

![DataSourceL类图][DataSourceL]  
接下来，我们分别看下三种数据源的实现方式。

### UnpooledDataSource 数据源创建过程 ###

上文提到，Mybatis 中的 dataSource 使用标准的 JDBC 数据源接口，因此 UnpooledDataSource 是实现了 DataSource 接口的，并重写了 DataSource 中的 getConnection 方法。  
看下 UnpooledDataSource 中 getConnection 方法实现：

public Connection getConnection() throws SQLException { 
        //调用 doGetConnection 获取连接
        return doGetConnection(username, password);
    }
    
    private Connection doGetConnection(String username, String password) throws SQLException { 
        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();
        //通过驱动代理类 DriverProxy 对象创建 connection
        Connection connection = DriverManager.getConnection(url, properties);
        //配置是否自动提交和事务隔离级别
        configureConnection(connection);
        return connection;
    }

UnpooledDataSource 的创建连接过程相对还是简单的。通过以下几步：

1.  获取数据库驱动配置。
2.  初始化数据库驱动。
3.  通过 DriverManager 获取连接。
4.  配置连接是否自动提交和事务隔离级别。

### PooledDataSource 数据源创建过程 ###

从 PooledDataSource 的代码中可以看出， PooledDataSource 同样继承 DataSource 接口，同时保存了一个 UnpooledDataSource 类的引用，还有一个 PoolState 对象用于保存空闲连接和活跃连接。我们看下获取连接代码：

public Connection getConnection() throws SQLException { 
        //获取真实连接的代理连接 proxyConnection
        return popConnection(dataSource.getUsername(), dataSource.getPassword()).getProxyConnection();
    }
    
    private PooledConnection popConnection(String username, String password) throws SQLException { 
        boolean countedWait = false;
        //PooledConnection 是 Connection 真实连接的代理类，用于关闭时特殊处理，下面会分析如何处理关闭情形
        PooledConnection conn = null;
        long t = System.currentTimeMillis();
        int localBadConnectionCount = 0;
    
        while (conn == null) { 
            //state 是 连接池状态，防止多线程操作，所以加锁
            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 { 
                    // 当活跃连接池数量小于最大活跃连接池数时，创建新的连接
                    if (state.activeConnections.size() < poolMaximumActiveConnections) { 
                        // Can create new connection
          //这里是用 UnpooledDataSource 对象来创建连接
                        conn = new PooledConnection(dataSource.getConnection(), this);
                        @SuppressWarnings("unused")
                        //used in logging, if enabled
                                Connection realConn = conn.getRealConnection();
                        if (log.isDebugEnabled()) { 
                            log.debug("Created connection " + conn.getRealHashCode() + ".");
                        }
                    } else { 
                        // 当活跃连接池数量等于最大活跃连接池数时，从活跃连接池的列表取第一个连接
                        PooledConnection oldestActiveConnection = state.activeConnections.get(0);
                        long longestCheckoutTime = oldestActiveConnection.getCheckoutTime();
                        //当池中连接被检出时间大于配置时间（单位毫秒），则重新new一个新连接，并将原连接置为不可用
                        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();
                        }
                        //设置连接类型编码，取url+username+password 三者拼接字符串的hashcode
                        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.");
                        }
                    }
                }
            }
    
        }
    
        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;
    }

PooledDataSource 的获取连接过程相对复杂一点，可以先看下流程图，简单描述了下上述代码的逻辑过程：  
![popConnection方法流程图][popConnection]  
接下来再通过文字描述下上述步骤：

1.  首先检查 PoolState 中的空闲连接池 idleConnections 中是否有可用连接。如果有，则直接从空闲连接池中取出一个连接使用。
2.  如果空闲连接池 idleConnections 中没有可用连接，那么继续检查 PoolState 中的活跃连接池 activeConnections 是否小于最大活跃连接池数。如果小于，则新建一个连接，新建连接是通过 `dataSource.getConnection()` 建立的，而 `dataSource` 是 `UnpooledDataSource` 对象，所以 `PooledDataSource` 中新建连接是通过 `UnpooledDataSource` 对象建立的，通过对象关联避免了重复代码。
3.  如果 PoolState 中的活跃连接池 activeConnections 不小于最大活跃连接池数，则从活跃连接池 activeConnections 中取出第一个连接，并检查是否超过连接池最大检出时间 poolMaximumCheckoutTime。如果未超过，则移除第一个活跃连接（最老的活跃连接），并新建一个连接。否则，继续等待获取连接，直到超过最大检出时间 poolMaximumCheckoutTime。
4.  if (conn != null) 如果连接不为空且可获得，则将获取的连接添加到活跃连接池中。否则，当重试超过一定次数的坏连接后，抛出 SQLException 异常。

**接下来，谈另一话题。正常我们执行 SQL 语句，都会在执行前打开连接，执行结束后关闭连接。在 `UnpooledDataSource` 中确实是关闭了真实连接，那 `PooledDataSource` 是如何关闭的呢？ 刚才说过，`PooledDataSource` 中 `getConnection` 返回的是代理连接 `proxyConnection`，重点就在这个代理类中，它通过代理真实连接对象，并对真实连接对象的调用方法进行了增强。代理类 `PooledConnection` 使用的是 JDK 动态代理，即实现了 InvocationHandler 接口，并重写了 invoke 方法，我们来看下代理类中是如何处理 `PooledDataSource` 数据源的关闭操作的。**

private static final String CLOSE = "close";
    
    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);
            }
        }
    }
    
    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++;
            }
        }
    }

从 `PooledConnection` 代理类中，我们看到了，代理类对连接的 `close` 方法做了特殊处理：

1.  当空闲连接池未达到最大空闲连接池数限制，则将真实连接加入空闲连接池。
2.  当空闲连接池数超过最大空闲连接池数限制，则将该真实连接关闭。  
    这样保证了 `PooledDataSource` 数据源中的连接复用，也是 `Mybatis` 中连接池的实现机制。

### JNDI 数据源创建过程 ###

Mybatis 创建 JNDI 数据源的方式只用到了一个工厂类 `JndiDataSourceFactory` ，JNDI 数据源创建方式和上面 UNPOOL 和 POOL 方式不同，JNDI 主要是通过启动上下文 `InitialContext` 和 `JNDI 配置文件`来完成。`InitialContext` 是执行命名操作的启动上下文，可以查找到配置文件中的 JNDI 配置：

public void setProperties(Properties properties) { 
        try { 
            InitialContext initCtx = null;
            //获取 JNDI 配置信息
            Properties env = getEnvProperties(properties);
            //初始化启动上下文
            if (env == null) { 
                initCtx = new InitialContext();
            } else { 
                initCtx = new InitialContext(env);
            }
            // 从配置文件中获取 dataSource
            if (properties.containsKey(INITIAL_CONTEXT)
                    && properties.containsKey(DATA_SOURCE)) { 
                Context ctx = (Context) initCtx.lookup(properties.getProperty(INITIAL_CONTEXT));
                dataSource = (DataSource) ctx.lookup(properties.getProperty(DATA_SOURCE));
            } else if (properties.containsKey(DATA_SOURCE)) { 
                dataSource = (DataSource) initCtx.lookup(properties.getProperty(DATA_SOURCE));
            }
        } catch (NamingException e) { 
            throw new DataSourceException("There was an error configuring JndiDataSourceTransactionPool. Cause: " + e, e);
        }

JNDI 获取数据源主要分三个步骤：

1.  首先获取 JNDI 配置信息。
2.  初始化启动上下文 initCtx。
3.  上下文调用 lookup 方法获取数据源 dataSource。

Mybatis 中的数据源类型和连接池实现原理就先介绍到这里，以备复习及其他小伙伴学习。

[DataSourceL]: /images/20230601/748b79307cea4fe3882704e8a0962993.png
[popConnection]: /images/20230601/79467d55462b48ad8ae9107ae7811b25.png