1 数据连接池介绍
2 连接池存储结构
3 连接池初始化
4 从连接池获取Connection
- 4.1 情况1 使用filter-chain模式来产生新的连接
- 4.2 情况2 直接获取getConnectionDirect()
5 通过Connection执行Sql工作机制
- 5.1 介绍
- 5.2 实现原理
6 关闭连接

本文概览：主要介绍了如下部分，（1）介绍了数据库连接池和Druid功能。(2)Druid相关操作，如下：

连接池初始化
从连接池获取Connection
通过connection执行sql
关闭connection

1 数据连接池介绍

1.1 数据库连接池引入

1、背景

通过JDBC直接访问数据库，使用这种方式，每一次执行sql，都需要创建新的连接，执行sql完成之后，还需要关闭连接。如下:

JDBC实例之mysql

当使用JDBC创建数据库连接时候，需要耗费很大的资源。如果在程序中，每次需要访问数据库时候，都进行数据库连接，那么势必会造成性能低下；同时，如果用户失误忘记释放数据库连接，会导致资源的浪费等。而数据库连接池就是解决该问题，通过管理连接池中的多个连接对象（connection），实现connection重复利用。从而，大大提高了数据库连接方面的性能。

2、连接池现状

目前存在多个开源的java数据库连接池，如DBCP、C3P0、Druid等，这些连接池都是基于JDK的线程池接口来实现，接口如下

public interface ConnectionPoolDataSource  extends CommonDataSource {
  // 常用
   PooledConnection getPooledConnection() throws SQLException;

   PooledConnection getPooledConnection(String user, String password)
    throws SQLException;
}

public interface ConnectionPoolDataSource extends CommonDataSource {

// 常用

PooledConnection getPooledConnection() throws SQLException;

PooledConnection getPooledConnection(String user, String password)

throws SQLException;

}

连接池中常涉及到两个参数有：

最小连接数：初始化时，创建该数目的connection放入连接池中。
最大连接数：允许创建connection的最大数值。当系统请求连接时候，且连接池中不存在空闲的连接，如果connection总数未超过最大连接数，那么连接池负责创建新的connection对象，并返回该对象；如果connection总数已经到达该最大连接数，那么连接池将用户请求转入等待队列。

1.2 Druid连接池介绍

Druid首先实现了一个数据库连接池的功能，这个功能和其他连接池没有什么差异，都是维护一个连接集合，包括数据库连接的创建、关闭等。但是Druid相比其他连接池，提供了监控功能，就通过filter-chain来实现各个filter的操作，实现对sql的监控，这正是Druid的价值所在。

Druid除了提供了一个数据库连接池功能。还具有如下功能：

（1）可以监控数据库访问性能，Druid内置提供了一个功能强大的StatFilter插件，能够详细统计SQL的执行性能，这对于线上分析数据库访问性能有帮助。

（2）数据库密码加密。直接把数据库密码写在配置文件中，这是不好的行为，容易导致安全问题。DruidDruiver和DruidDataSource都支持PasswordCallback。

（3）SQL执行日志，Druid提供了不同的LogFilter，能够支持Common-Logging、Log4j和JdkLog，你可以按需要选择相应的LogFilter，监控你应用的数据库访问情况。

（4）自定义过滤器。扩展JDBC，如果你要对JDBC层有编程的需求，可以通过Druid提供的Filter机制，很方便编写JDBC层的扩展插件。

1.3 Druid与ConnectionPoolDataSource比较

实现了jvax.sql.ConnectionPoolDataSource的接口中第一个函数接口，对于第二个函数不进行支持

   @Override
    public PooledConnection getPooledConnection() throws SQLException {
        return getConnection(maxWait);
    }

    // 德鲁伊不支持这个带参数接口函数的实现
    @Override
    public PooledConnection getPooledConnection(String user, String password) throws SQLException {
        throw new UnsupportedOperationException("Not supported by DruidDataSource");
    }

@Override

public PooledConnection getPooledConnection() throws SQLException {

return getConnection(maxWait);

}

// 德鲁伊不支持这个带参数接口函数的实现

@Override

public PooledConnection getPooledConnection(String user, String password) throws SQLException {

throw new UnsupportedOperationException("Not supported by DruidDataSource");

}

2 连接池存储结构

连接池中的连接存储：Conetions和activeConnections两个数组。

 // 存放的是：从连接池获取连接时，从这个数组中获取
  private volatile DruidConnectionHolder[] connections;
  
  // 存放的是：正在使用的连接connection
  protected final Map<DruidPooledConnection, Object> activeConnections

// 存放的是：从连接池获取连接时，从这个数组中获取

private volatile DruidConnectionHolder[] connections;

// 存放的是：正在使用的连接connection

protected final Map<DruidPooledConnection, Object> activeConnections

有以下说明：

连接池中保存的元素是什么？是DruidConnectionHolder（保存了JDK的connection对象）。
连接池中每一个连接，用完之后必须调用close()。在close中通过recyCle（）把连接从activeConections集合中重新移动到connections[]，达到重复利用连接的目的。

1、关于DruidDataSource的activeConnections

在activeConections中添加连接，是在DruidDataSource#getConnection时进行获取的。
删除activeConnections中连接，在关闭连接DruidPooledConneciton#close中使用DruidDataSource#recycle，如下：

 protected void recycle(DruidPooledConnection pooledConnection) throws SQLException {
    ...
  }

protected void recycle(DruidPooledConnection pooledConnection) throws SQLException {

...

}

所以，使用DruidDataSouce连接池获取数据库连接时，在使用完这个connection之后，如果没有进行close操作，那么activeConnections就不会删除，只会增加。此时因为线程池中动态创建连接的线程CreateConnectionThread中判断一个条件就是

 if (activeCount + poolingCount >= maxActive) {...}

1	if (activeCount + poolingCount >= maxActive) {...}

此时一个场景就是poolingCount=0，但是activeCount的线程数目是已经达到了maxAcive，那么此时线程池就会发现无法再提供连接的故障。

为了解决这个问题，所以在init函数中启动了一个DestroyConnectionThread线程来完成这件事情。但是需要用户设置一个removeAbandoned变量才可以生效。

2、connections

（1） connections添加一个连接

在init()初始化时进行增加；
在createAndStartCreatorThread中进行增加

（2）connections删除一个连接

在getConnectionInternal中获取连接时，进行删除。通过这个函数获取的连接已经被关闭，那么此时在调用这个函数的getConnectionDirect中会进行判断这个获取连接是否关闭，如果关闭，再次通过getConnectionInternal进行获取,如下代码:

public DruidPooledConnection getConnectionDirect(long maxWaitMillis) throws SQLException {
        for (;;) {
            DruidPooledConnection poolalbeConnection = getConnectionInternal(maxWaitMillis);

            if (isTestOnBorrow()) {
             // 1.进行校验，失败则删除相应的activeConections中连接
                boolean validate = testConnectionInternal(poolalbeConnection.getConnection());
                if (!validate) {
                    if (LOG.isDebugEnabled()) {
                        LOG.debug("skip not validate connection.");
                    }

                    Connection realConnection = poolalbeConnection.getConnection();
                    discardConnection(realConnection);
                    continue;
                }
            } else {
                Connection realConnection = poolalbeConnection.getConnection();               
               //2.进行校验，失败则删除相应的activeConections中连接
                if (realConnection.isClosed()) {
                    discardConnection(null); // 传入null，避免重复关闭
                    continue;
 
......               }
}

public DruidPooledConnection getConnectionDirect(long maxWaitMillis) throws SQLException {

for (;;) {

DruidPooledConnection poolalbeConnection = getConnectionInternal(maxWaitMillis);

if (isTestOnBorrow()) {

// 1.进行校验，失败则删除相应的activeConections中连接

boolean validate = testConnectionInternal(poolalbeConnection.getConnection());

if (!validate) {

if (LOG.isDebugEnabled()) {

LOG.debug("skip not validate connection.");

}

Connection realConnection = poolalbeConnection.getConnection();

discardConnection(realConnection);

continue;

}

} else {

Connection realConnection = poolalbeConnection.getConnection();

//2.进行校验，失败则删除相应的activeConections中连接

if (realConnection.isClosed()) {

discardConnection(null); // 传入null，避免重复关闭

continue;

...... }

}

3 连接池初始化

通过init函数进行初始化连接池。这个函数在getConection()时进行调用。在执行init函数时，首先会初始化存放连接的数组conections，然后启动两个线程

（1）CreateConnectionThread，动态在connections中添加连接

（2）DestroyConnectionThread，动态删除actiConnections中很久不用的连接。

3.1 通过init来初始化

1、通过init函数完成初始化

public void init() throws SQLException {

     // 1. 初始化连接
     for (int i = 0, size = getInitialSize(); i < size; ++i) {
                    Connection conn = createPhysicalConnection();
                    DruidConnectionHolder holder = new DruidConnectionHolder(this, conn);
                    connections[poolingCount++] = holder;
      }
     ........
     
     // 2.记录线程持仓统计信息的线程
     createAndLogThread();
    // 3. 启动CreateConnectionThread，动态在connections中添加连
       createAndStartCreatorThread();  
    // 4. 启动 DestroyConnectionThread，动态删除actiConnections中很久不用的连接。
        createAndStartDestroyThread();
}

public void init() throws SQLException {

// 1. 初始化连接

for (int i = 0, size = getInitialSize(); i < size; ++i) {

Connection conn = createPhysicalConnection();

DruidConnectionHolder holder = new DruidConnectionHolder(this, conn);

connections[poolingCount++] = holder;

}

........

// 2.记录线程持仓统计信息的线程

createAndLogThread();

// 3. 启动CreateConnectionThread，动态在connections中添加连

createAndStartCreatorThread();

// 4. 启动 DestroyConnectionThread，动态删除actiConnections中很久不用的连接。

createAndStartDestroyThread();

}

2、初始化时机

（1）方式1：在配置文件中

   <bean id="dataSource" class="com.alibaba.druid.pool.DruidDataSource" init-method="init" destroy-method="close">
    ...
   </bean>

...

</bean>

（2）方式2：通过第一次执行getConnection

public DruidPooledConnection getConnection(long maxWaitMillis) throws SQLException {
       //在init中 执行初始化.如果此时已经初始化了连接池，那么此时就直接返回。
         init();

if (filters.size() > 0) {
            FilterChainImpl filterChain = new FilterChainImpl(this);
            return filterChain.dataSource_connect(this, maxWaitMillis);
        } else {
            return getConnectionDirect(maxWaitMillis);
        }
    }

public DruidPooledConnection getConnection(long maxWaitMillis) throws SQLException {

//在init中执行初始化.如果此时已经初始化了连接池，那么此时就直接返回。

init();

if (filters.size() > 0) {

FilterChainImpl filterChain = new FilterChainImpl(this);

return filterChain.dataSource_connect(this, maxWaitMillis);

} else {

return getConnectionDirect(maxWaitMillis);

}

3.2 动态创建连接CreateConnectionThread

该线程负责动态的创建连接，维护连接数在initSize到maxActive之间。如下代码说明了在连接池动态创建一个连接的时机：

public class CreateConnectionThread extends Thread {
  public void run() {

            try {
                  // 1.必须存在线程等待，才创建连接
                 // notEcmptyWaitThreadCount值为：
                // 执行takeLast（）或者pollLast（long nanos）操作,发现 poolingCount = 0时等待的线程的个数。
                    if (poolingCount >= notEmptyWaitThreadCount) {
                        empty.await();  // 进行阻塞
                    }

                    // 2. 防止创建超过maxActive数量的连接
                    // activeCount是正在运行的使用连接个数；
                    // poolingCount是线程池中连接个数，初始化时该值为initialSize，最大为maxActive；
                    if (activeCount + poolingCount >= maxActive) {
                        
                        empty.await();   // 进行阻塞
                        continue;
                    }
                
                
                Connection connection = null;
                try {
                    // 3.通过JDBC创建一个连接
                    connection = createPhysicalConnection();
                } catch (SQLException e) {
                ....
               }
                
                // 4.放入到connections[]数组中。并且增加pollingCount的值
                 put(connection);
 }
}

public class CreateConnectionThread extends Thread {

public void run() {

try {

// 1.必须存在线程等待，才创建连接

// notEcmptyWaitThreadCount值为：

// 执行takeLast（）或者pollLast（long nanos）操作,发现 poolingCount = 0时等待的线程的个数。

if (poolingCount >= notEmptyWaitThreadCount) {

empty.await(); // 进行阻塞

}

// 2. 防止创建超过maxActive数量的连接

// activeCount是正在运行的使用连接个数；

// poolingCount是线程池中连接个数，初始化时该值为initialSize，最大为maxActive；

if (activeCount + poolingCount >= maxActive) {

empty.await(); // 进行阻塞

continue;

}

Connection connection = null;

try {

// 3.通过JDBC创建一个连接

connection = createPhysicalConnection();

} catch (SQLException e) {

....

}

// 4.放入到connections[]数组中。并且增加pollingCount的值

put(connection);

}

3.3 动态删除连接DestroyConnectionThread

该线程的作用就是：关闭连接池connnections[]空闲连接和连接池activeConections[]中超时运行连接。

1、创建并启动线程代码如下：

protected void createAndStartDestroyThread() {
        destoryTask = new DestroyTask();
        // 需要自己配置
        if (destroyScheduler != null) {
            long period = timeBetweenEvictionRunsMillis;
            if (period <= 0) {
                period = 1000;
            }
            destroySchedulerFuture = destroyScheduler.scheduleAtFixedRate(destoryTask, period, period,
                                                                          TimeUnit.MILLISECONDS);
            initedLatch.countDown();
            return;
        }

        // 默认情况下,通过DestroyTask来完成。
        String threadName = "Druid-ConnectionPool-Destroy-" + System.identityHashCode(this);
        destroyConnectionThread = new DestroyConnectionThread(threadName);
        destroyConnectionThread.start();
    }

protected void createAndStartDestroyThread() {

destoryTask = new DestroyTask();

// 需要自己配置

if (destroyScheduler != null) {

long period = timeBetweenEvictionRunsMillis;

if (period <= 0) {

period = 1000;

}

destroySchedulerFuture = destroyScheduler.scheduleAtFixedRate(destoryTask, period, period,

TimeUnit.MILLISECONDS);

initedLatch.countDown();

return;

}

// 默认情况下,通过DestroyTask来完成。

String threadName = "Druid-ConnectionPool-Destroy-" + System.identityHashCode(this);

destroyConnectionThread = new DestroyConnectionThread(threadName);

destroyConnectionThread.start();

}

2、通过DetroyTask来实现，包括两部分：

关闭连接池connnections[]中空闲时间超过一个阈值的连接
关闭连接池activeConneection[] 运行超过一个阈值的连接

 public class DestroyTask implements Runnable {

        @Override
        public void run() {
            // 1.关闭 连接池connnections[]中 空闲时间超过一个阈值的连接
            shrink(true);

            // 2.removeAbandoned属性值为ture时
            if (isRemoveAbandoned()) {
                // 在activeConnections中查找运行时间超过removeAbandonedTimeoutMillis的连接，然后通过JDBC的close方法进行关闭
                removeAbandoned();
            }
        }

    }

public class DestroyTask implements Runnable {

@Override

public void run() {

// 1.关闭连接池connnections[]中空闲时间超过一个阈值的连接

shrink(true);

// 2.removeAbandoned属性值为ture时

if (isRemoveAbandoned()) {

// 在activeConnections中查找运行时间超过removeAbandonedTimeoutMillis的连接，然后通过JDBC的close方法进行关闭

removeAbandoned();

}

（1）shrink(true)

用于关闭connecitons[]中空闲连接数，涉及到属性有：

minIdle：最小空闲连接数
minEvictableIdleTimeMillis 空闲时间的阈值，超过改阈值需要进行关闭该空闲连接

public void shrink(boolean checkTime) {
        .....
        try { 
            final int checkCount = poolingCount - minIdle;
            final long currentTimeMillis = System.currentTimeMillis();
            // 1.连接数是否大于最小空闲连接数，如果大于，则需要处理多余的空闲congtion，否则不进行处理
            for (int i = 0; i < checkCount; ++i) {
                DruidConnectionHolder connection = connections[i];
                if (checkTime) {
                    // 2.选取空闲时间大于minEvictableIdleTimeMillis的connection
                    long idleMillis = currentTimeMillis - connection.getLastActiveTimeMillis();
                    if (idleMillis >= minEvictableIdleTimeMillis) {
                        evictList.add(connection);
                    } else {
                        break;
                    }
                } else {
                    evictList.add(connection);
                }
            }
         .....
        // 3.通过JDBC关闭多余空闲connnection
        for (DruidConnectionHolder item : evictList) {
            Connection connection = item.getConnection();
            JdbcUtils.close(connection);
            destroyCount.incrementAndGet();
        }
    }

public void shrink(boolean checkTime) {

.....

try {

final int checkCount = poolingCount - minIdle;

final long currentTimeMillis = System.currentTimeMillis();

// 1.连接数是否大于最小空闲连接数，如果大于，则需要处理多余的空闲congtion，否则不进行处理

for (int i = 0; i < checkCount; ++i) {

DruidConnectionHolder connection = connections[i];

if (checkTime) {

// 2.选取空闲时间大于minEvictableIdleTimeMillis的connection

long idleMillis = currentTimeMillis - connection.getLastActiveTimeMillis();

if (idleMillis >= minEvictableIdleTimeMillis) {

evictList.add(connection);

} else {

break;

}

} else {

evictList.add(connection);

}

.....

// 3.通过JDBC关闭多余空闲connnection

for (DruidConnectionHolder item : evictList) {

Connection connection = item.getConnection();

JdbcUtils.close(connection);

destroyCount.incrementAndGet();

}

（2）removeAbandoned()

处理activeConenctions[]中运行超时连接，涉及到属性有：

removeAbandoned。是否需要处理activeConections超时连接：true表示处理；false表示不处理。
removeAbandonedTimeoutMillis。运行超时的阈值

public int removeAbandoned() {
     Iterator<DruidPooledConnection> iter = activeConnections.keySet().iterator();
     for (; iter.hasNext();) {
        // 1.获取运行超时的connection   
        long timeMillis = (currrentNanos - pooledConnection.getConnectedTimeNano()) / (1000 * 1000);
        if (timeMillis >= removeAbandonedTimeoutMillis) {
            ....
            abandonedList.add(pooledConnection);  
        }
     }
     
     for (DruidPooledConnection pooledConnection : abandonedList) {
        // 2.通过JDBC关闭该连接
        JdbcUtils.close(pooledConnection);
     }
     
 }

public int removeAbandoned() {

Iterator<DruidPooledConnection> iter = activeConnections.keySet().iterator();

for (; iter.hasNext();) {

// 1.获取运行超时的connection

long timeMillis = (currrentNanos - pooledConnection.getConnectedTimeNano()) / (1000 * 1000);

if (timeMillis >= removeAbandonedTimeoutMillis) {

....

abandonedList.add(pooledConnection);

}

for (DruidPooledConnection pooledConnection : abandonedList) {

// 2.通过JDBC关闭该连接

JdbcUtils.close(pooledConnection);

}

4 从连接池获取Connection

通过两种方式可以获取连接

直接获取
通过filter-chain获取

@Override
 public DruidPooledConnection getConnection() throws SQLException {
       // 默认会设置一个最大连接时间，这个最大设置时间需要在配置DruidDataSource时进行配置。
        return getConnection(maxWait);
  }

  public DruidPooledConnection getConnection(long maxWaitMillis) throws SQLException {
        init();
        // 1.情况1 包含过滤器，则通过filter-chain模式来产生新的连接
        if (filters.size() > 0) {
            FilterChainImpl filterChain = new FilterChainImpl(this);
            return filterChain.dataSource_connect(this, maxWaitMillis);
        } else {

         // 2.情况2 不包含过滤器
            return getConnectionDirect(maxWaitMillis);
        }
    }

@Override

public DruidPooledConnection getConnection() throws SQLException {

// 默认会设置一个最大连接时间，这个最大设置时间需要在配置DruidDataSource时进行配置。

return getConnection(maxWait);

}

public DruidPooledConnection getConnection(long maxWaitMillis) throws SQLException {

init();

// 1.情况1 包含过滤器，则通过filter-chain模式来产生新的连接

if (filters.size() > 0) {

FilterChainImpl filterChain = new FilterChainImpl(this);

return filterChain.dataSource_connect(this, maxWaitMillis);

} else {

// 2.情况2 不包含过滤器

return getConnectionDirect(maxWaitMillis);

}

4.1 情况1 使用filter-chain模式来产生新的连接

查看FilterChainImpl#dataSource_connect实现代码

@Override
public DruidPooledConnection dataSource_connect(DruidDataSource dataSource, long maxWaitMillis) throws SQLException {
      // 1. 通过一个pos来表示执行到filters[pos]的过滤器，直到执行完filters的所有过滤器，才执行下一步
        if (this.pos < filterSize) {
            DruidPooledConnection conn = nextFilter().dataSource_getConnection(this, dataSource, maxWaitMillis);
            return conn;
        }

      // 2. 通过情况2中getConnectionDirect来从连接池获取一个连接。
        return dataSource.getConnectionDirect(maxWaitMillis);
}

@Override

public DruidPooledConnection dataSource_connect(DruidDataSource dataSource, long maxWaitMillis) throws SQLException {

// 1. 通过一个pos来表示执行到filters[pos]的过滤器，直到执行完filters的所有过滤器，才执行下一步

if (this.pos < filterSize) {

DruidPooledConnection conn = nextFilter().dataSource_getConnection(this, dataSource, maxWaitMillis);

return conn;

}

// 2. 通过情况2中getConnectionDirect来从连接池获取一个连接。

return dataSource.getConnectionDirect(maxWaitMillis);

}

这里使用filter的目的，就在于在filter可以做一些操作，如下两个例子

以LogFilter#dataSource_getConnection为例

public DruidPooledConnection dataSource_getConnection(FilterChain chain, DruidDataSource dataSource,
                                                          long maxWaitMillis) throws SQLException {

        //1. 继续执行filterchain#dataSource_connect，来执行下一个filter#dataSource_getConnection（每一个过滤都需要一个这步操作）
        DruidPooledConnection conn = chain.dataSource_connect(dataSource, maxWaitMillis);

            //2. 这个过滤器需要完成的操作
        ConnectionProxy connection = (ConnectionProxy) conn.getConnectionHolder().getConnection();
        if (connectionConnectAfterLogEnable && isConnectionLogEnabled()) {
            connectionLog("{conn-" + connection.getId() + "} pool-connect");
        }

   // 3.返回conn
        return conn;
    }

public DruidPooledConnection dataSource_getConnection(FilterChain chain, DruidDataSource dataSource,

long maxWaitMillis) throws SQLException {

//1. 继续执行filterchain#dataSource_connect，来执行下一个filter#dataSource_getConnection（每一个过滤都需要一个这步操作）

DruidPooledConnection conn = chain.dataSource_connect(dataSource, maxWaitMillis);

//2. 这个过滤器需要完成的操作

ConnectionProxy connection = (ConnectionProxy) conn.getConnectionHolder().getConnection();

if (connectionConnectAfterLogEnable && isConnectionLogEnabled()) {

connectionLog("{conn-" + connection.getId() + "} pool-connect");

}

// 3.返回conn

return conn;

}

以StatFilter#dataSource_getConnection为例

 @Override
    public DruidPooledConnection dataSource_getConnection(FilterChain chain, DruidDataSource dataSource,
                                                          long maxWaitMillis) throws SQLException {
    //1. 继续执行filterchain#dataSource_connect，来执行下一个filter#dataSource_getConnection（每一个过滤都需要一个这步操作）
        DruidPooledConnection conn = chain.dataSource_connect(dataSource, maxWaitMillis);

    //2. 这个过滤器需要完成的操作
        if (conn != null) {
            conn.setConnectedTimeNano();
            StatFilterContext.getInstance().pool_connection_open();
        }

 // 3.返回conn
  return conn;
    }

@Override

public DruidPooledConnection dataSource_getConnection(FilterChain chain, DruidDataSource dataSource,

long maxWaitMillis) throws SQLException {

//1. 继续执行filterchain#dataSource_connect，来执行下一个filter#dataSource_getConnection（每一个过滤都需要一个这步操作）

DruidPooledConnection conn = chain.dataSource_connect(dataSource, maxWaitMillis);

//2. 这个过滤器需要完成的操作

if (conn != null) {

conn.setConnectedTimeNano();

StatFilterContext.getInstance().pool_connection_open();

}

// 3.返回conn

return conn;

}

由上，filter操作模板就是包括3步：

 DruidPooledConnection dataSource_connect(DruidDataSource dataSource, long maxWaitMillis) throws SQLException{
 //1. 继续执行filterchain#dataSource_connect，来执行下一个filter#dataSource_getConnection（每一个过滤都需要一个这步操作）
 DruidPooledConnection conn = chain.dataSource_connect(dataSource, maxWaitMillis);

   //2. 这个过滤器需要完成的操作
   .......
   .......

 // 3.返回conn
  return conn;
}

DruidPooledConnection dataSource_connect(DruidDataSource dataSource, long maxWaitMillis) throws SQLException{

//1. 继续执行filterchain#dataSource_connect，来执行下一个filter#dataSource_getConnection（每一个过滤都需要一个这步操作）

DruidPooledConnection conn = chain.dataSource_connect(dataSource, maxWaitMillis);

//2. 这个过滤器需要完成的操作

.......

// 3.返回conn

return conn;

}

所以，假设DruidDataSources中filters包括三个filter：filter1,filter2,filter3，每一个dataSource_getConnection代码模板为：

{
  //1. 继续执行filterchain#dataSource_connect，来执行下一个filter#dataSource_getConnection（每一个过滤都需要一个这步操作） 
 DruidPooledConnection conn = chain.dataSource_connect(dataSource, maxWaitMillis); 

   //2. 这个过滤器需要完成的操作 
   System.out.pringln(执行当前类名字：xxx.class)
   .......

 // 3.返回conn
  return conn;
}

{

//1. 继续执行filterchain#dataSource_connect，来执行下一个filter#dataSource_getConnection（每一个过滤都需要一个这步操作）

DruidPooledConnection conn = chain.dataSource_connect(dataSource, maxWaitMillis);

//2. 这个过滤器需要完成的操作

System.out.pringln(执行当前类名字：xxx.class)

.......

// 3.返回conn

return conn;

}

则在DruidDataSource执行getConnection（）之后，此时打印结果就是：(类似于递归，先走到底，然后再返回依次执行)

执行当前类名字：filter3
执行当前类名字：filter2
执行当前类名字：filter1

执行当前类名字：filter3

执行当前类名字：filter2

执行当前类名字：filter1

4.2 情况2 直接获取getConnectionDirect()

在getConnectionDirect中获取流程

public DruidPooledConnection getConnectionDirect(long maxWaitMillis) throws SQLException {
        for (;;) {
           // 1. 从connections[]获取连接，并从connections[]中删除这个连接，然后在activeConnections中增加这个连接
            DruidPooledConnection poolalbeConnection = getConnectionInternal(maxWaitMillis);
           
          // 2.判断连接是否可以使用，如果不可用：执行discardConnection将这个连接从connection执行connetion.close。然后继续执行continue，通过getConnectionInternal来获取连接。
         // 2.1  测试connetions是否有效可用
         if (isTestOnBorrow()) {
             // 通过执行sql校验connnection是否有效
            boolean validate = testConnectionInternal(poolableConnection.getConnection());
             
          } else {
              // 2.2 测试是否close
                Connection realConnection = poolalbeConnection.getConnection(); 
             
            // 2.3 对超过一个空闲时间阈值的连接，进行测试该连接是否有效可用。
               if (isTestWhileIdle()) {
                  long timeBetweenEvictionRunsMillis = this.getTimeBetweenEvictionRunsMillis();
                    if (timeBetweenEvictionRunsMillis <= 0) {
                        timeBetweenEvictionRunsMillis = DEFAULT_TIME_BETWEEN_EVICTION_RUNS_MILLIS;
                    }
                    // 超过一个空闲等待时间的连接，进行测试conneciton是否有效可用
                    if (idleMillis >= timeBetweenEvictionRunsMillis) {
                        boolean validate = testConnectionInternal(poolableConnection.getConnection())
               }
          }

         
        // 3.如果可以，则执行返回获取的连接      
}

public DruidPooledConnection getConnectionDirect(long maxWaitMillis) throws SQLException {

for (;;) {

// 1. 从connections[]获取连接，并从connections[]中删除这个连接，然后在activeConnections中增加这个连接

DruidPooledConnection poolalbeConnection = getConnectionInternal(maxWaitMillis);

// 2.判断连接是否可以使用，如果不可用：执行discardConnection将这个连接从connection执行connetion.close。然后继续执行continue，通过getConnectionInternal来获取连接。

// 2.1 测试connetions是否有效可用

if (isTestOnBorrow()) {

// 通过执行sql校验connnection是否有效

boolean validate = testConnectionInternal(poolableConnection.getConnection());

} else {

// 2.2 测试是否close

Connection realConnection = poolalbeConnection.getConnection();

// 2.3 对超过一个空闲时间阈值的连接，进行测试该连接是否有效可用。

if (isTestWhileIdle()) {

long timeBetweenEvictionRunsMillis = this.getTimeBetweenEvictionRunsMillis();

if (timeBetweenEvictionRunsMillis <= 0) {

timeBetweenEvictionRunsMillis = DEFAULT_TIME_BETWEEN_EVICTION_RUNS_MILLIS;

}

// 超过一个空闲等待时间的连接，进行测试conneciton是否有效可用

if (idleMillis >= timeBetweenEvictionRunsMillis) {

boolean validate = testConnectionInternal(poolableConnection.getConnection())

}

// 3.如果可以，则执行返回获取的连接

}

1、getConnectionInternal(maxWaitMillis);

在DruidDataSource中的如下代码：

private DruidPooledConnection getConnectionInternal(long maxWait) throws SQLException {
          ....
        // 加锁。获取一个连接池元素时,需要添加锁，然后从connections[]数组中获取连接，
        loock.lock();
        ...   
        // 如果设置了最大等待时间maxWait，就
          if (maxWait > 0) {
                holder = pollLast(nanos);
            } else {
                holder = takeLast();
            }
    ....
         // 解锁
         lock.unlock
    ...
}

private DruidPooledConnection getConnectionInternal(long maxWait) throws SQLException {

....

// 加锁。获取一个连接池元素时,需要添加锁，然后从connections[]数组中获取连接，

loock.lock();

...

// 如果设置了最大等待时间maxWait，就

if (maxWait > 0) {

holder = pollLast(nanos);

} else {

holder = takeLast();

}

....

// 解锁

lock.unlock

...

}

（1）lock

这也可以解释为什么在获取连接时，需要增加超时时间，因为lock是阻塞的。 在获取连接时，如果设置了超时，则获取连接时间如果超时则会抛出异常（参考pollLast(nanos)中的代码）

(2) pollLast(nanos)

private DruidConnectionHolder pollLast(long nanos) throws InterruptedException, SQLException {
    // 连接池connections[]没有可用连接时，阻塞等待最大等待时间
      estimate = notEmpty.awaitNanos(estimate);
    ...
    // 获取连接
    decrementPoolingCount();
    DruidConnectionHolder last = connections[poolingCount];
    connections[poolingCount] = null;
   
}

private DruidConnectionHolder pollLast(long nanos) throws InterruptedException, SQLException {

// 连接池connections[]没有可用连接时，阻塞等待最大等待时间

estimate = notEmpty.awaitNanos(estimate);

...

// 获取连接

decrementPoolingCount();

DruidConnectionHolder last = connections[poolingCount];

connections[poolingCount] = null;

}

（3）takeLast

DruidConnectionHolder takeLast() throws InterruptedException, SQLException {
    
    // 连接池connections[]没有可用连接时，一直会阻塞等待
       notEmpty.await();
    ...
    // 获取连接
    decrementPoolingCount();
    DruidConnectionHolder last = connections[poolingCount];
    connections[poolingCount] = null;
    
}

DruidConnectionHolder takeLast() throws InterruptedException, SQLException {

// 连接池connections[]没有可用连接时，一直会阻塞等待

notEmpty.await();

...

// 获取连接

decrementPoolingCount();

DruidConnectionHolder last = connections[poolingCount];

connections[poolingCount] = null;

}

2、testConnectionInternal(poolableConnection.getConnection())

通过使用获取的conneciton执行一个测试sql来校验是否生效。

protected boolean testConnectionInternal(Connection conn) {
    ....
         Statement stmt = null;
            ResultSet rset = null;
            try {
                stmt = conn.createStatement();
                if (getValidationQueryTimeout() > 0) {
                    // 1.设置执行sql超时时间
                    stmt.setQueryTimeout(validationQueryTimeout);
                }
                // 2.执行测试sql
                rset = stmt.executeQuery(validationQuery);
                if (!rset.next()) {
                    return false;
                }
            } finally {
                JdbcUtils.close(rset);
                JdbcUtils.close(stmt);
            }

         return true
 }

protected boolean testConnectionInternal(Connection conn) {

....

Statement stmt = null;

ResultSet rset = null;

try {

stmt = conn.createStatement();

if (getValidationQueryTimeout() > 0) {

// 1.设置执行sql超时时间

stmt.setQueryTimeout(validationQueryTimeout);

}

// 2.执行测试sql

rset = stmt.executeQuery(validationQuery);

if (!rset.next()) {

return false;

}

} finally {

JdbcUtils.close(rset);

JdbcUtils.close(stmt);

}

return true

}

3、isTestOnBorrow()和isTestWhileIdle()比较

二者只能有一个为true，如果都为true，优先选择isTestOnBorrow()。

(1)共同点

都是校验connection可用

(2)区别

isTestWhileIdle()只针对空闲时间超过一定时间的连接。而testConnectionInternal是针对所有连接

4、涉及到属性有：

maxWait，获取连接时的最大等待时间
testOnBorrow，是否校验connection有效可用。
validationQuery，在testOnBorrow为true情况下，获取连接后，通过执行该sql校验connection是否正常
validationQueryTimeout，在testOnBorrow为ture情况下，执行validationQeury时，超时时间，默认不进行超时设置
testWhileIdle，超过一个空闲等待时间的连接，进行测试conneciton是否有效可用。在testOnBorrow为false情况下使用。
timeBetweenEvictionRunsMillis，空闲等待时间，testWhileIdle=true的时候使用

5 通过Connection执行Sql工作机制

druid最大价值就在于提供监控功能。通过filter-chain来实现各个filter的操作（如自定义了一个FilterEventAdapter,那么就可以实现对执行sql的监控）。

使用filter执行sql的流程：

（1）第一步初始化连接池

（2）第二步获取DruidPooledPreparedStatement，这个类中的statement的值有两种情况。

（3）第三步执行PreparedStatementProxyImpl

5.1 介绍

在“从连接池获取连接”小节中实现ConnectionPoolDataSource#getPooledConnection()查询连接时，使用filter-chain模式来产生新的连接，类似于递归。filter-chain除了上述获取连接，还有查询、更新等数据库操作。以执行查询为例，查看failterChainimpl的statement_executeQuery操作为例，如下代码：

public ResultSetProxy statement_executeQuery(StatementProxy statement, String sql) throws SQLException {
        if (this.pos < filterSize) {
            return nextFilter().statement_executeQuery(this, statement, sql);
        }

        ResultSet resultSet = statement.getRawObject().executeQuery(sql);

        if (resultSet == null) {
            return null;
        }

        return wrap(statement, resultSet);
    }

public ResultSetProxy statement_executeQuery(StatementProxy statement, String sql) throws SQLException {

if (this.pos < filterSize) {

return nextFilter().statement_executeQuery(this, statement, sql);

}

ResultSet resultSet = statement.getRawObject().executeQuery(sql);

if (resultSet == null) {

return null;

}

return wrap(statement, resultSet);

}

5.2 实现原理

通过druid连接池得到DruidPooledConnection，通过这个连接获取到DruidPooledStatement.

1、初始化连接

在初始化连接池时，需要创建连接，分为如下两种情况：

如果filters不为空，则在连接池中中DruidConnectionHolder包含的connectio的类型就是ConnectionProxyImpl。
如果filters为空，则在连接池中DruidConnectionHolder包含的connectio的类型就是Connection。

public Connection createPhysicalConnection(String url, Properties info) throws SQLException {
        Connection conn;
        if (getProxyFilters().size() == 0) {
            conn = getDriver().connect(url, info);
        } else {
            conn = new FilterChainImpl(this).connection_connect(info);
        }

        createCount.incrementAndGet();

        return conn;
    }

public Connection createPhysicalConnection(String url, Properties info) throws SQLException {

Connection conn;

if (getProxyFilters().size() == 0) {

conn = getDriver().connect(url, info);

} else {

conn = new FilterChainImpl(this).connection_connect(info);

}

createCount.incrementAndGet();

return conn;

}

2、获取DruidPooledPreparedStatement

（1）从初始化连接池中获取连接DruidPooledConnection,

（2）通过DruidPooledConnection来获取DruidPooledPreparedStatement。这里分为两种情况：

包含过滤器（即DruidDataSource中filters不为空）
不包含过滤器（即DruidDataSource中filters为空）

查看DruidPooledConnection#prepareStatement操作

public PreparedStatement prepareStatement(String sql) throws SQLException {
        ........
        ........

        if (stmtHolder == null) {
            try {
            // 1.这里分根据conn的类型分为两种情况：
            //（1）如果conn是Connection类型，则此时就是preparedStatement类型
            //（2）如果conn是ConnectionProxyImpl，此时就是PreparedStatementProxyImpl
                stmtHolder = new PreparedStatementHolder(key, conn.prepareStatement(sql));
                holder.getDataSource().incrementPreparedStatementCount();
            } catch (SQLException ex) {
                handleException(ex);
            }
        }
       // 2.DruidPooledPrepardStatement就是封装了一个stament，这个statement由stmtholder得到，对于statement类型有两种，如第一步。
        DruidPooledPreparedStatement rtnVal = new DruidPooledPreparedStatement(this, stmtHolder);

        holder.addTrace(rtnVal);

        return rtnVal;
    }

public PreparedStatement prepareStatement(String sql) throws SQLException {

........

if (stmtHolder == null) {

try {

// 1.这里分根据conn的类型分为两种情况：

//（1）如果conn是Connection类型，则此时就是preparedStatement类型

//（2）如果conn是ConnectionProxyImpl，此时就是PreparedStatementProxyImpl

stmtHolder = new PreparedStatementHolder(key, conn.prepareStatement(sql));

holder.getDataSource().incrementPreparedStatementCount();

} catch (SQLException ex) {

handleException(ex);

}

// 2.DruidPooledPrepardStatement就是封装了一个stament，这个statement由stmtholder得到，对于statement类型有两种，如第一步。

DruidPooledPreparedStatement rtnVal = new DruidPooledPreparedStatement(this, stmtHolder);

holder.addTrace(rtnVal);

return rtnVal;

}

3、执行PreparedStatementProxyImpl实现监控功能

如果包含过滤器就通过PreparedStatementProxyImpl来执行sql，查看执行函数

 public boolean execute() throws SQLException {
        ......
        // 通过filterChain来执行
        firstResultSet = createChain().preparedStatement_execute(this);
        return firstResultSet;
  }

public boolean execute() throws SQLException {

......

// 通过filterChain来执行

firstResultSet = createChain().preparedStatement_execute(this);

return firstResultSet;

}

4、 如何确认是否包含过虑器？

就是看DruidDataSource中filters是否为空，即看用户在配置DruidDataSource时，是否配置了filters。

6 关闭连接

通过关闭连接实现回收连接，达到可重复利用connection的目的。在执行close操作时也分为：使用过滤器和不使用过滤器两种，如下代码

public void close() throws SQLException { 
       .....
       List<Filter> filters = dataSource.getProxyFilters();
        // 1.使用过滤器
        if (filters.size() > 0) {
            FilterChainImpl filterChain = new FilterChainImpl(dataSource);
            filterChain.dataSource_recycle(this);
        } else {
           // 2.直接回收，不使用过滤器
            recycle();
        }
}

public void close() throws SQLException {

.....

List<Filter> filters = dataSource.getProxyFilters();

// 1.使用过滤器

if (filters.size() > 0) {

FilterChainImpl filterChain = new FilterChainImpl(dataSource);

filterChain.dataSource_recycle(this);

} else {

// 2.直接回收，不使用过滤器

recycle();

}

在执行DruidPooledConnections#close，执行如下recycle，分为是否回收连接两种情况：

public void recycle() throws SQLException {
        if (this.disable) {
            return;
        }

        DruidConnectionHolder holder = this.holder;
        if (holder == null) {
            if (dupCloseLogEnable) {
                LOG.error("dup close");
            }
            return;
        }
// 1.情况1 abandoned为false，表示还没有通过JDBC关闭该连接，此时可以进行回收连接，重复利用
        if (!this.abandoned) {
            DruidAbstractDataSource dataSource = holder.getDataSource();
            dataSource.recycle(this);
        }

// 2.情况2 abandoned为true，表示已经通过JDBC关闭了该连接，此时不进行回收线程
        this.holder = null;
        conn = null;
        transactionInfo = null;
        closed = true;
    }

public void recycle() throws SQLException {

if (this.disable) {

return;

}

DruidConnectionHolder holder = this.holder;

if (holder == null) {

if (dupCloseLogEnable) {

LOG.error("dup close");

}

return;

}

// 1.情况1 abandoned为false，表示还没有通过JDBC关闭该连接，此时可以进行回收连接，重复利用

if (!this.abandoned) {

DruidAbstractDataSource dataSource = holder.getDataSource();

dataSource.recycle(this);

}

// 2.情况2 abandoned为true，表示已经通过JDBC关闭了该连接，此时不进行回收线程

this.holder = null;

conn = null;

transactionInfo = null;

closed = true;

}

1、设置DruidPooledConnections的abandoned属性为true的时机

在DestroyConnectionThread线程执行DestroyTask时，在通过removeAbandoned()关闭超时运行连接时，设置了abndoned属性为true：

public class DestroyTask implements Runnable {

        @Override
        public void run() {
            shrink(true);

            if (isRemoveAbandoned()) {
                // 通过JDBC关闭超时连接，并设置连接的abandoned属性为true
                removeAbandoned();
            }
        }
}

public class DestroyTask implements Runnable {

@Override

public void run() {

shrink(true);

if (isRemoveAbandoned()) {

// 通过JDBC关闭超时连接，并设置连接的abandoned属性为true

removeAbandoned();

}

removeAbandoned()执行逻辑如下，通过JDBC关闭超时连接，并设置连接的abandoned属性为true

public int removeAbandoned() {
        ....
         for (DruidPooledConnection pooledConnection : abandonedList) {
                synchronized (pooledConnection) {
                    if (pooledConnection.isDisable()) {
                        continue;
                    }
                }
                
                JdbcUtils.close(pooledConnection);
                // 设置连接的abandoned属性为true
                pooledConnection.abandond();
         }
       ....
    }

public int removeAbandoned() {

....

for (DruidPooledConnection pooledConnection : abandonedList) {

synchronized (pooledConnection) {

if (pooledConnection.isDisable()) {

continue;

}

JdbcUtils.close(pooledConnection);

// 设置连接的abandoned属性为true

pooledConnection.abandond();

}

....

}

2、dataSource.recycle(this)

作用：将conneciton重新放到connections[]中。

  /**
     * 回收连接，重新把连接放回connections[]中
     */
    protected void recycle(DruidPooledConnection pooledConnection) throws SQLException {
        
       ....
       final Connection physicalConnection = holder.getConnection();
       ....
       // 1、校验回收的连接是否可以有效使用
       if (testOnReturn) {
          boolean validate = testConnectionInternal(physicalConnection);
          ...
       }
        
        // 2、将连接重新放到connecitons[]中
        putLast(holder, lastActiveTimeMillis);
    }

/**

* 回收连接，重新把连接放回connections[]中

protected void recycle(DruidPooledConnection pooledConnection) throws SQLException {

....

final Connection physicalConnection = holder.getConnection();

....

// 1、校验回收的连接是否可以有效使用

if (testOnReturn) {

boolean validate = testConnectionInternal(physicalConnection);

...

}

// 2、将连接重新放到connecitons[]中

putLast(holder, lastActiveTimeMillis);

}

3、涉及到相关属性有

testOnReturn 是否需要对回收的连接，检查有效可用

（全文完）

Heart.Think.Do

Druid连接池之DruidDataSource

1 数据连接池介绍

1.1 数据库连接池引入

1.2 Druid连接池介绍

1.3 Druid与ConnectionPoolDataSource比较

2 连接池存储结构

3 连接池初始化

3.1 通过init来初始化

3.2 动态创建连接CreateConnectionThread

3.3 动态删除连接DestroyConnectionThread

4 从连接池获取Connection

4.1 情况1 使用filter-chain模式来产生新的连接

4.2 情况2 直接获取getConnectionDirect()

5 通过Connection执行Sql工作机制

5.1 介绍

5.2 实现原理

6 关闭连接

Related Posts:

分类&标签