设置每次统计前重置所有统计变量
dataSourceStat.setResetStatEnable(this.resetStatEnable);
声明的三个DruidConnectionHolder数组代表什么
// 记录存放数据库链接
connections = new DruidConnectionHolder[maxActive];
// 记录被剔除的数据库链接
evictConnections = new DruidConnectionHolder[maxActive];
// 记录检测后存活的链接
keepAliveConnections = new DruidConnectionHolder[maxActive];
启动时异步创建线程池
// 需要在properties里配置createScheduler 并且 asyncInit =true,默认asyncInit=false
if (createScheduler != null && asyncInit) {
for (int i = 0; i < initialSize; ++i) {
// 此处逻辑和下面同步差不多
submitCreateTask(true);
}
}
启动时同步创建连接池
默认initialSize=0 需要DruidDataSource.setInitialSize(?),并且设置的数值是不能超过maxActive(默认8)的,否则报错.
为什么需要initialSize,我觉得是为了应对流量比较大的应用,可以考虑在启动时候就开启连接池,否则会遇到第一条sql特别慢的情况,造成不必要的slowsql统计.
dataSourceStat.setResetStatEnable(this.resetStatEnable);
connections = new DruidConnectionHolder[maxActive];
evictConnections = new DruidConnectionHolder[maxActive];
keepAliveConnections = new DruidConnectionHolder[maxActive];
SQLException connectError = null;
// 异步初始化 需要properties里配置createScheduler和asyncInit设置为true,这个是兼容老版本,异步
if (createScheduler != null && asyncInit) {
for (int i = 0; i < initialSize; ++i) {
submitCreateTask(true);
}
} else if (!asyncInit) {
// initialSize 默认0 指的是初始化时候的大小,如果指定会创建链接对象.否则会再调用的时候创建
while (poolingCount < initialSize) {
try {
// 创建物理链接 封装了 datasource的connection信息
PhysicalConnectionInfo pyConnectInfo = createPhysicalConnection();
// 为每个connection设置对应的holder connectionId是key 记录了每个链接上一次链接的信息
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);
}
}
}
// peak代表峰值 供统计参考
if (poolingCount > 0) {
poolingPeak = poolingCount;
poolingPeakTime = System.currentTimeMillis();
}
}
// 创建统计连接池状态的线程,间隔是 timeBetweenLogStatsMillis
createAndLogThread();
// 异步 初始化从连接池里获取链接的线程 = 从连接池拿物理链接并且put到map的holder中 保证不会被gc
createAndStartCreatorThread();
// 异步 初始化从连接池里销毁的线程 间隔为 timeBetweenEvictionRunsMillis
createAndStartDestroyThread();
// 堵塞 createAndStartCreatorThread和createAndStartDestroyThread 都会countDown 也就是说等这2个线程初始化完成才会做这个
initedLatch.await();
init = true;
创建连接池
同步和异步创建最后都会走到这
public void run() {
// 使latch的值减1，如果减到了0，则会唤醒所有等待在这个latch上的线程
initedLatch.countDown();
System.out.println("CreateConnectionThread initedLatch countDown " + initedLatch.getCount());
long lastDiscardCount = 0;
int errorCount = 0;
for (; ; ) {
// addLast
try {
lock.lockInterruptibly();
} catch (InterruptedException e2) {
// 中断死循环
break;
}
long discardCount = DruidDataSource.this.discardCount;
boolean discardChanged = discardCount - lastDiscardCount > 0;
lastDiscardCount = discardCount;
try {
// true代表不允许创建
boolean emptyWait = true;
// 开始各种创建连接池的判断
if (createError != null && poolingCount == 0 && !discardChanged) {
// 如果创建时没发生错误 并且 当前池子里也没有链接 并且 丢弃链接的数量大于0
emptyWait = false;
}
// 一般asyncInit=false 不会异步任务创建. initialSize也是0
if (emptyWait && asyncInit && createCount < initialSize) {
// 如果是异步创建的方式 并且 创建的数量小于初始化的数量,那么也允许创建连接
emptyWait = false;
}
// 初始化时候核心逻辑,有一个false则允许创建
if (emptyWait) {
// 当前连接池里可用的链接数 大于等于 消费等待的线程数,初始化时 poolingCount >= notEmptyWaitThreadCount 代表有需求者要连接
if (poolingCount >= notEmptyWaitThreadCount
// 如果开启了keepAlive 并且活动的连接池+可用的连接池数 小于 最小的空闲连接数
&& (!(keepAlive && activeCount + poolingCount < minIdle))
// 如果没有连续的错误
&& !isFailContinuous()) {
// 堵塞 不能创建
System.out.println("堵塞 empty.await()");
empty.await();
}
System.out.println("empty 唤醒");
// 防止创建超过maxActive数量的连接
if (activeCount + poolingCount >= maxActive) {
// 堵塞 不能创建
System.out.println("堵塞生产者 活动链接 + 当前可用链接 大于等于 maxActive");
empty.await();
continue;
}
}
} catch (InterruptedException e) {
lastCreateError = e;
lastErrorTimeMillis = System.currentTimeMillis();
if ((!closing) && (!closed)) {
LOG.error("create connection Thread Interrupted, url: " + jdbcUrl, e);
}
break;
} finally {
lock.unlock();
}
PhysicalConnectionInfo connection = null;
// 创建链接
try {
connection = createPhysicalConnection();
} catch (SQLException e) {
LOG.error("create connection SQLException, url: " + jdbcUrl + ", errorCode " + e.getErrorCode()
+ ", state " + e.getSQLState(), e);
errorCount++;
// timeBetweenConnectErrorMillis = 500ms
if (errorCount > connectionErrorRetryAttempts && timeBetweenConnectErrorMillis > 0) {
// fail over retry attempts
setFailContinuous(true);
if (failFast) {
lock.lock();
try {
notEmpty.signalAll();
} finally {
lock.unlock();
}
}
if (breakAfterAcquireFailure) {
break;
}
try {
Thread.sleep(timeBetweenConnectErrorMillis);
} catch (InterruptedException interruptEx) {
break;
}
}
} catch (RuntimeException e) {
LOG.error("create connection RuntimeException", e);
setFailContinuous(true);
continue;
} catch (Error e) {
LOG.error("create connection Error", e);
setFailContinuous(true);
break;
}
// 如果创建错误 那么connection 是空,因为catch了错误
if (connection == null) {
continue;
}
boolean result = put(connection);
if (!result) {
JdbcUtils.close(connection.getPhysicalConnection());
LOG.info("put physical connection to pool failed.");
}
errorCount = 0; // reset errorCount
if (closing || closed) {
break;
}
}
}
}
