Day02 Druid 如何初始化、添加、销毁、获取连接

在执行 DruidDataSource#init() 方法时，会初始化存放连接的数组conections，然后启动两个线程

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

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

初始化连接：

while (poolingCount < initialSize) {
  try {
      PhysicalConnectionInfo pyConnectInfo = createPhysicalConnection();
      DruidConnectionHolder holder = new DruidConnectionHolder(this, pyConnectInfo);
      connections[poolingCount++] = holder;
  } catch (SQLException ex) {
      LOG.error("init datasource error, url: " + this.getUrl(), ex);
      if (initExceptionThrow) {
          connectError = ex;
          break;
      } else {
          Thread.sleep(3000);
      }
  }
}

CreateConnectionThread 是一个守护线程，当有线程等待时，便会通过 JDBC 创建不超过 maxActive 数量的数据库连接，然后将创建的连接放入 connections 数组。

public class CreateConnectionThread extends Thread {

        public CreateConnectionThread(String name){
            super(name);
            this.setDaemon(true);
        }

        public void run() {
        	for (;;) {
                ...
                ...
                // 通过 JDBC 创建一个数据库连接
                connection = createPhysicalConnection();
                ...
                ...
                // 将创建的连接放入 connections 数组
                boolean result = put(connection);
            }
        }
        
}

DestroyConnectionThread 也是一个守护线程

    public class DestroyConnectionThread extends Thread {

        public DestroyConnectionThread(String name){
            super(name);
            this.setDaemon(true);
        }

        public void run() {
            initedLatch.countDown();

            for (;;) {
                // 从前面开始删除
                try {
                    if (closed || closing) {
                        break;
                    }

                    if (timeBetweenEvictionRunsMillis > 0) {
                        Thread.sleep(timeBetweenEvictionRunsMillis);
                    } else {
                        Thread.sleep(1000); //
                    }

                    if (Thread.interrupted()) {
                        break;
                    }
					// 执行删除数据库连接的任务
                    destroyTask.run();
                } catch (InterruptedException e) {
                    break;
                }
            }
        }

    }

DestroyTask ：

    public class DestroyTask implements Runnable {
        public DestroyTask() {}
        @Override
        public void run() {
            // 关闭连接池 connnections[] 中空闲时间超过一个阈值的连接
            shrink(true, keepAlive);

            if (isRemoveAbandoned()) {
                // 移除 activeConnections[] 中运行超 removeAbandonedTimeoutMillis 的连接，并通过 JDBC 来 close 掉对应的数据库连接
                removeAbandoned();
            }
        }

    }

    
    public DruidPooledConnection getConnection(long maxWaitMillis) throws SQLException {
        // 初始化工作
        init();

        if (filters.size() > 0) {
            FilterChainImpl filterChain = new FilterChainImpl(this);
            // 走过滤器拿数据库连接，详看FilterChainImpl#dataSource_connect
            return filterChain.dataSource_connect(this, maxWaitMillis);
        } else {
            // 直接从 connections[] 获取连接,详见 holder = takeLast();
            return getConnectionDirect(maxWaitMillis);
        }
    }

getConnectionDirect()：

public DruidPooledConnection getConnectionDirect(long maxWaitMillis) throws SQLException {
        int notFullTimeoutRetryCnt = 0;
        for (;;) {
            // handle notFullTimeoutRetry
            DruidPooledConnection poolableConnection;
            try {
                // 从connections[]数组中获取连接，maxWaitMillis————拿连接的最大等待时间
                poolableConnection = getConnectionInternal(maxWaitMillis);
            } catch (GetConnectionTimeoutException ex) {
                if (notFullTimeoutRetryCnt <= this.notFullTimeoutRetryCount && !isFull()) {
                    notFullTimeoutRetryCnt++;
                    if (LOG.isWarnEnabled()) {
                        LOG.warn("get connection timeout retry : " + notFullTimeoutRetryCnt);
                    }
                    continue;
                }
                throw ex;
            }

            if (testOnBorrow) {
                // 用获取的连接执行一个测试sql来校验是否生效
                boolean validate = testConnectionInternal(poolableConnection.holder, poolableConnection.conn);
                ...
            }
            ...
        }
    ...
}