diff --git a/fine-druid/src/com/fr/third/alibaba/druid/pool/DruidDataSource.java b/fine-druid/src/com/fr/third/alibaba/druid/pool/DruidDataSource.java index f3e71fef9..edae35120 100644 --- a/fine-druid/src/com/fr/third/alibaba/druid/pool/DruidDataSource.java +++ b/fine-druid/src/com/fr/third/alibaba/druid/pool/DruidDataSource.java @@ -15,6 +15,43 @@ */ package com.fr.third.alibaba.druid.pool; +import java.io.Closeable; +import java.security.AccessController; +import java.security.PrivilegedAction; +import java.sql.Connection; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.ConcurrentModificationException; +import java.util.Date; +import java.util.HashMap; +import java.util.Iterator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.ServiceLoader; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +import javax.management.JMException; +import javax.management.MBeanRegistration; +import javax.management.MBeanServer; +import javax.management.ObjectName; +import javax.naming.NamingException; +import javax.naming.Reference; +import javax.naming.Referenceable; +import javax.naming.StringRefAddr; +import javax.sql.ConnectionEvent; +import javax.sql.ConnectionEventListener; +import javax.sql.ConnectionPoolDataSource; +import javax.sql.PooledConnection; + import com.fr.third.alibaba.druid.Constants; import com.fr.third.alibaba.druid.TransactionTimeoutException; import com.fr.third.alibaba.druid.VERSION; @@ -58,121 +95,84 @@ import com.fr.third.alibaba.druid.util.Utils; import com.fr.third.alibaba.druid.wall.WallFilter; import com.fr.third.alibaba.druid.wall.WallProviderStatValue; -import javax.management.JMException; -import javax.management.MBeanRegistration; -import javax.management.MBeanServer; -import javax.management.ObjectName; -import javax.naming.NamingException; -import javax.naming.Reference; -import javax.naming.Referenceable; -import javax.naming.StringRefAddr; -import javax.sql.ConnectionEvent; -import javax.sql.ConnectionEventListener; -import javax.sql.ConnectionPoolDataSource; -import javax.sql.PooledConnection; -import java.io.Closeable; -import java.security.AccessController; -import java.security.PrivilegedAction; -import java.sql.Connection; -import java.sql.SQLException; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.ConcurrentModificationException; -import java.util.Date; -import java.util.HashMap; -import java.util.Iterator; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.Properties; -import java.util.ServiceLoader; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLong; -import java.util.concurrent.locks.Lock; -import java.util.concurrent.locks.ReentrantLock; - /** * @author ljw [ljw2083@alibaba-inc.com] * @author wenshao [szujobs@hotmail.com] */ public class DruidDataSource extends DruidAbstractDataSource implements DruidDataSourceMBean, ManagedDataSource, Referenceable, Closeable, Cloneable, ConnectionPoolDataSource, MBeanRegistration { - private final static Log LOG = LogFactory.getLog(DruidDataSource.class); + private final static Log LOG = LogFactory.getLog(DruidDataSource.class); - private static final long serialVersionUID = 1L; + private static final long serialVersionUID = 1L; // stats - private final AtomicLong recycleErrorCount = new AtomicLong(); - private long connectCount = 0L; - private long closeCount = 0L; - private final AtomicLong connectErrorCount = new AtomicLong(); - private long recycleCount = 0L; - private long removeAbandonedCount = 0L; - private long notEmptyWaitCount = 0L; - private long notEmptySignalCount = 0L; - private long notEmptyWaitNanos = 0L; - - private int keepAliveCheckCount = 0; - - private int activePeak = 0; - private long activePeakTime = 0; - private int poolingPeak = 0; - private long poolingPeakTime = 0; + private final AtomicLong recycleErrorCount = new AtomicLong(); + private long connectCount = 0L; + private long closeCount = 0L; + private final AtomicLong connectErrorCount = new AtomicLong(); + private long recycleCount = 0L; + private long removeAbandonedCount = 0L; + private long notEmptyWaitCount = 0L; + private long notEmptySignalCount = 0L; + private long notEmptyWaitNanos = 0L; + + private int keepAliveCheckCount = 0; + + private int activePeak = 0; + private long activePeakTime = 0; + private int poolingPeak = 0; + private long poolingPeakTime = 0; // store private volatile DruidConnectionHolder[] connections; - private int poolingCount = 0; - private int activeCount = 0; - private long discardCount = 0; - private int notEmptyWaitThreadCount = 0; - private int notEmptyWaitThreadPeak = 0; + private int poolingCount = 0; + private int activeCount = 0; + private long discardCount = 0; + private int notEmptyWaitThreadCount = 0; + private int notEmptyWaitThreadPeak = 0; // - private DruidConnectionHolder[] evictConnections; - private DruidConnectionHolder[] keepAliveConnections; + private DruidConnectionHolder[] evictConnections; + private DruidConnectionHolder[] keepAliveConnections; // threads - private ScheduledFuture destroySchedulerFuture; - private DestroyTask destroyTask; + private ScheduledFuture destroySchedulerFuture; + private DestroyTask destroyTask; - private CreateConnectionThread createConnectionThread; - private DestroyConnectionThread destroyConnectionThread; - private LogStatsThread logStatsThread; - private int createTaskCount; + private CreateConnectionThread createConnectionThread; + private DestroyConnectionThread destroyConnectionThread; + private LogStatsThread logStatsThread; + private int createTaskCount; - private final CountDownLatch initedLatch = new CountDownLatch(2); + private CountDownLatch initedLatch = new CountDownLatch(2); - private volatile boolean enable = true; + private volatile boolean enable = true; - private boolean resetStatEnable = true; - private final AtomicLong resetCount = new AtomicLong(); + private boolean resetStatEnable = true; + private final AtomicLong resetCount = new AtomicLong(); - private String initStackTrace; + private String initStackTrace; - private volatile boolean closed = false; - private long closeTimeMillis = -1L; + private volatile boolean closed = false; + private long closeTimeMillis = -1L; - protected JdbcDataSourceStat dataSourceStat; + protected JdbcDataSourceStat dataSourceStat; - private boolean useGlobalDataSourceStat = false; + private boolean useGlobalDataSourceStat = false; - private boolean mbeanRegistered = false; + private boolean mbeanRegistered = false; - public static ThreadLocal waitNanosLocal = new ThreadLocal(); + public static ThreadLocal waitNanosLocal = new ThreadLocal(); - private boolean logDifferentThread = true; + private boolean logDifferentThread = true; - private volatile boolean keepAlive = false; + private volatile boolean keepAlive = false; - private static final boolean logActiveCount = "true".equals(System.getProperty("frDruidLogActiveCount")); - - public DruidDataSource(){ + public DruidDataSource() { this(false); } - public DruidDataSource(boolean fairLock){ + public DruidDataSource(boolean fairLock) { super(fairLock); configFromPropety(System.getProperties()); @@ -603,6 +603,35 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat this.connectProperties = properties; } + private synchronized void createThreadChange() { + String threadName = "Druid-ConnectionPool-Create-" + System.identityHashCode(this) + this.getUrl(); + createConnectionThread = new CreateConnectionThread(threadName); + createConnectionThread.setStarted(false); + String destroyName = "Druid-ConnectionPool-Destroy-" + System.identityHashCode(this) + this.getUrl(); + if (destroyConnectionThread != null) { + if (!destroyConnectionThread.isInterrupted()) { + destroyConnectionThread.interrupt(); + } + } + destroyConnectionThread = new DestroyConnectionThread(destroyName); + destroyConnectionThread.setStarted(false); + initedLatch = new CountDownLatch(2); + } + + private void checkThread() throws SQLException { + if (!createConnectionThread.isStarted() && !destroyConnectionThread.isStarted()) { + createConnectionThread.setStarted(true); + createConnectionThread.start(); + destroyConnectionThread.setStarted(true); + destroyConnectionThread.start(); + try { + initedLatch.await(); + } catch (InterruptedException e) { + throw new SQLException(e.getMessage(), e); + } + } + } + public void init() throws SQLException { if (inited) { return; @@ -646,7 +675,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } if (JdbcConstants.MYSQL.equals(this.dbType) || // - JdbcConstants.MARIADB.equals(this.dbType)) { + JdbcConstants.MARIADB.equals(this.dbType)) { boolean cacheServerConfigurationSet = false; if (this.connectProperties.containsKey("cacheServerConfiguration")) { cacheServerConfigurationSet = true; @@ -806,7 +835,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat period = 1000; } destroySchedulerFuture = destroyScheduler.scheduleAtFixedRate(destroyTask, period, period, - TimeUnit.MILLISECONDS); + TimeUnit.MILLISECONDS); initedLatch.countDown(); return; } @@ -818,7 +847,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat protected void createAndStartCreatorThread() { if (createScheduler == null) { - String threadName = "Druid-ConnectionPool-Create-" + System.identityHashCode(this); + String threadName = "Druid-ConnectionPool-Create-" + System.identityHashCode(this) + this.getUrl(); createConnectionThread = new CreateConnectionThread(threadName); createConnectionThread.start(); return; @@ -929,7 +958,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat if (driver.getMajorVersion() < 10) { throw new SQLException("not support oracle driver " + driver.getMajorVersion() + "." - + driver.getMinorVersion()); + + driver.getMinorVersion()); } if (driver.getMajorVersion() == 10 && isUseOracleImplicitCache()) { @@ -966,7 +995,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat if (query instanceof SQLSelectQueryBlock) { if (((SQLSelectQueryBlock) query).getFrom() == null) { LOG.error("invalid oracle validationQuery. " + validationQuery + ", may should be : " + validationQuery - + " FROM DUAL"); + + " FROM DUAL"); } } } @@ -995,7 +1024,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat if (query instanceof SQLSelectQueryBlock) { if (((SQLSelectQueryBlock) query).getFrom() == null) { LOG.error("invalid db2 validationQuery. " + validationQuery + ", may should be : " + validationQuery - + " FROM SYSDUMMY"); + + " FROM SYSDUMMY"); } } } @@ -1003,14 +1032,14 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat private void initValidConnectionChecker() { String realDriverClassName = driver.getClass().getName(); if (realDriverClassName.equals(JdbcConstants.MYSQL_DRIVER) // - || realDriverClassName.equals(JdbcConstants.MYSQL_DRIVER_6)) { + || realDriverClassName.equals(JdbcConstants.MYSQL_DRIVER_6)) { this.validConnectionChecker = new MySqlValidConnectionChecker(); } else if (realDriverClassName.equals(JdbcConstants.ORACLE_DRIVER) || realDriverClassName.equals(JdbcConstants.ORACLE_DRIVER2)) { this.validConnectionChecker = new OracleValidConnectionChecker(); } else if (realDriverClassName.equals(JdbcConstants.SQL_SERVER_DRIVER) - || realDriverClassName.equals(JdbcConstants.SQL_SERVER_DRIVER_SQLJDBC4) - || realDriverClassName.equals(JdbcConstants.SQL_SERVER_DRIVER_JTDS)) { + || realDriverClassName.equals(JdbcConstants.SQL_SERVER_DRIVER_SQLJDBC4) + || realDriverClassName.equals(JdbcConstants.SQL_SERVER_DRIVER_JTDS)) { this.validConnectionChecker = new MSSQLValidConnectionChecker(); } else if (realDriverClassName.equals(JdbcConstants.POSTGRESQL_DRIVER)) { this.validConnectionChecker = new PGValidConnectionChecker(); @@ -1028,7 +1057,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat String realDriverClassName = driver.getClass().getName(); if (realDriverClassName.equals(JdbcConstants.MYSQL_DRIVER) // - || realDriverClassName.equals(JdbcConstants.MYSQL_DRIVER_6)) { + || realDriverClassName.equals(JdbcConstants.MYSQL_DRIVER_6)) { this.exceptionSorter = new MySqlExceptionSorter(); } else if (realDriverClassName.equals(JdbcConstants.ORACLE_DRIVER) || realDriverClassName.equals(JdbcConstants.ORACLE_DRIVER2)) { @@ -1056,7 +1085,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat public DruidPooledConnection getConnection(long maxWaitMillis) throws SQLException { init(); - + checkThread(); if (filters.size() > 0) { FilterChainImpl filterChain = new FilterChainImpl(this); return filterChain.dataSource_connect(this, maxWaitMillis); @@ -1077,7 +1106,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat public DruidPooledConnection getConnectionDirect(long maxWaitMillis) throws SQLException { int notFullTimeoutRetryCnt = 0; - for (;;) { + for (; ; ) { // handle notFullTimeoutRetry DruidPooledConnection poolableConnection; try { @@ -1167,7 +1196,6 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat lock.lock(); try { activeCount--; - logActiveCount(false); discardCount++; if (activeCount <= minIdle) { @@ -1205,7 +1233,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat if (notEmptyWaitThreadCount >= maxWaitThreadCount) { connectErrorCount.incrementAndGet(); throw new SQLException("maxWaitThreadCount " + maxWaitThreadCount + ", current wait Thread count " - + lock.getQueueLength()); + + lock.getQueueLength()); } } @@ -1219,7 +1247,6 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat if (holder != null) { activeCount++; - logActiveCount(true); if (activeCount > activePeak) { activePeak = activeCount; activePeakTime = System.currentTimeMillis(); @@ -1238,7 +1265,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat if (holder == null) { Throwable throwable = this.createError; if (throwable != null) { - if(throwable instanceof SQLException){ + if (throwable instanceof SQLException) { throw (SQLException) throwable; } throw new GetConnectionTimeoutException(throwable.getMessage(), throwable); @@ -1251,19 +1278,21 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat .append(", active " + activeCount)// .append(", maxActive " + maxActive)// ; - - List sqlList = this.getDataSourceStat().getRuningSqlList(); - for (int i = 0; i < sqlList.size(); ++i) { - if (i != 0) { - buf.append('\n'); - } else { - buf.append(", "); + JdbcDataSourceStat sourceStat = this.getDataSourceStat(); + if (sourceStat != null) { + List sqlList = sourceStat.getRuningSqlList(); + for (int i = 0; i < sqlList.size(); ++i) { + if (i != 0) { + buf.append('\n'); + } else { + buf.append(", "); + } + JdbcSqlStatValue sql = sqlList.get(i); + buf.append("runningSqlCount "); + buf.append(sql.getRunningCount()); + buf.append(" : "); + buf.append(sql.getSql()); } - JdbcSqlStatValue sql = sqlList.get(i); - buf.append("runningSqlCount "); - buf.append(sql.getRunningCount()); - buf.append(" : "); - buf.append(sql.getSql()); } String errorMessage = buf.toString(); throw new GetConnectionTimeoutException(errorMessage); @@ -1345,9 +1374,9 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } if (logDifferentThread // - && (!isAsyncCloseConnectionEnable()) // - && pooledConnection.getOwnerThread() != Thread.currentThread()// - ) { + && (!isAsyncCloseConnectionEnable()) // + && pooledConnection.getOwnerThread() != Thread.currentThread()// + ) { LOG.warn("get/close not same thread"); } @@ -1409,7 +1438,6 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat lock.lock(); try { activeCount--; - logActiveCount(false); closeCount++; } finally { lock.unlock(); @@ -1428,7 +1456,6 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat lock.lockInterruptibly(); try { activeCount--; - logActiveCount(false); closeCount++; result = putLast(holder, lastActiveTimeMillis); @@ -1546,7 +1573,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat @Override public Object run() { ObjectName objectName = DruidDataSourceStatManager.addDataSource(DruidDataSource.this, - DruidDataSource.this.name); + DruidDataSource.this.name); DruidDataSource.this.setObjectName(objectName); DruidDataSource.this.mbeanRegistered = true; @@ -1636,7 +1663,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat private DruidConnectionHolder pollLast(long nanos) throws InterruptedException, SQLException { long estimate = nanos; - for (;;) { + for (; ; ) { if (poolingCount == 0) { emptySignal(); // send signal to CreateThread create connection @@ -1657,8 +1684,8 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat try { long startEstimate = estimate; estimate = notEmpty.awaitNanos(estimate); // signal by - // recycle or - // creator + // recycle or + // creator notEmptyWaitCount++; notEmptyWaitNanos += (startEstimate - estimate); @@ -1929,7 +1956,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat createTaskCount--; if (poolingCount + createTaskCount < notEmptyWaitThreadCount // - && activeCount + poolingCount + createTaskCount < maxActive) { + && activeCount + poolingCount + createTaskCount < maxActive) { emptySignal(); } } @@ -1949,7 +1976,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } private void runInternal() { - for (;;) { + for (; ; ) { // addLast lock.lock(); @@ -2017,9 +2044,6 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat LOG.error("create connection error", e); // unknow fatal exception setFailContinuous(true); - if (breakAfterAcquireFailure) { - break; - } continue; } catch (Error e) { lock.lock(); @@ -2049,8 +2073,9 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } public class CreateConnectionThread extends Thread { + private volatile boolean started = true; - public CreateConnectionThread(String name){ + public CreateConnectionThread(String name) { super(name); this.setDaemon(true); } @@ -2060,7 +2085,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat long lastDiscardCount = 0; int errorCount = 0; - for (;;) { + for (; ; ) { // addLast try { lock.lockInterruptibly(); @@ -2108,7 +2133,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat setFailContinuous(false); } catch (SQLException e) { LOG.error("create connection error, url: " + jdbcUrl + ", errorCode " + e.getErrorCode() - + ", state " + e.getSQLState(), e); + + ", state " + e.getSQLState(), e); errorCount++; if (errorCount > connectionErrorRetryAttempts && timeBetweenConnectErrorMillis > 0) { @@ -2132,13 +2157,12 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } catch (InterruptedException interruptEx) { break; } + DruidDataSource.this.createThreadChange(); + break; } } catch (RuntimeException e) { LOG.error("create connection error", e); setFailContinuous(true); - if (breakAfterAcquireFailure) { - break; - } continue; } catch (Error e) { LOG.error("create connection error", e); @@ -2159,11 +2183,20 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat errorCount = 0; // reset errorCount } } + + public boolean isStarted() { + return started; + } + + public void setStarted(boolean started) { + this.started = started; + } } public class DestroyConnectionThread extends Thread { + private volatile boolean started = true; - public DestroyConnectionThread(String name){ + public DestroyConnectionThread(String name) { super(name); this.setDaemon(true); } @@ -2171,7 +2204,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat public void run() { initedLatch.countDown(); - for (;;) { + for (; ; ) { // 从前面开始删除 try { if (closed) { @@ -2195,6 +2228,13 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } } + public boolean isStarted() { + return started; + } + + public void setStarted(boolean started) { + this.started = started; + } } public class DestroyTask implements Runnable { @@ -2212,14 +2252,14 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat public class LogStatsThread extends Thread { - public LogStatsThread(String name){ + public LogStatsThread(String name) { super(name); this.setDaemon(true); } public void run() { try { - for (;;) { + for (; ; ) { try { logStats(); } catch (Exception e) { @@ -2245,7 +2285,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat try { Iterator iter = activeConnections.keySet().iterator(); - for (; iter.hasNext();) { + for (; iter.hasNext(); ) { DruidPooledConnection pooledConnection = iter.next(); if (pooledConnection.isRunning()) { @@ -2297,7 +2337,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } buf.append("ownerThread current state is " + pooledConnection.getOwnerThread().getState() - + ", current stackTrace\n"); + + ", current stackTrace\n"); trace = pooledConnection.getOwnerThread().getStackTrace(); for (int i = 0; i < trace.length; i++) { buf.append("\tat "); @@ -2313,7 +2353,9 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat return removeCount; } - /** Instance key */ + /** + * Instance key + */ protected String instanceKey = null; public Reference getReference() throws NamingException { @@ -2947,7 +2989,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } if (this.statLogger != null - && (this.statLogger.getClass() == iface || DruidDataSourceStatLogger.class == iface)) { + && (this.statLogger.getClass() == iface || DruidDataSourceStatLogger.class == iface)) { return true; } @@ -2963,7 +3005,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } if (this.statLogger != null - && (this.statLogger.getClass() == iface || DruidDataSourceStatLogger.class == iface)) { + && (this.statLogger.getClass() == iface || DruidDataSourceStatLogger.class == iface)) { return (T) statLogger; } @@ -3007,7 +3049,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } int fillCount = 0; - for (;;) { + for (; ; ) { try { lock.lockInterruptibly(); } catch (InterruptedException e) { @@ -3129,10 +3171,4 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat public boolean isClosed() { return this.closed; } - - private void logActiveCount(boolean isIncrease){ - if (logActiveCount){ - LOG.debug("activeCount " + (isIncrease ? "increase" : "decrease") + "current count is : " + activeCount, new Throwable()); - } - } }