From 914346455a09c79111f238a6b923ab4f641fb906 Mon Sep 17 00:00:00 2001 From: abel Date: Mon, 6 Aug 2018 10:41:13 +0800 Subject: [PATCH 1/3] =?UTF-8?q?druid=E6=96=AD=E7=BA=BF=E9=87=8D=E8=BF=9E?= =?UTF-8?q?=E6=9C=BA=E5=88=B6=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../alibaba/druid/pool/DruidDataSource.java | 218 +++++++++++------- 1 file changed, 132 insertions(+), 86 deletions(-) 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 341aeb540..dc0c3b042 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 @@ -35,6 +35,7 @@ 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; @@ -100,78 +101,78 @@ import com.fr.third.alibaba.druid.wall.WallProviderStatValue; */ 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; - public DruidDataSource(){ + public DruidDataSource() { this(false); } - public DruidDataSource(boolean fairLock){ + public DruidDataSource(boolean fairLock) { super(fairLock); configFromPropety(System.getProperties()); @@ -602,6 +603,30 @@ 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(); + 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; @@ -645,7 +670,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; @@ -805,7 +830,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat period = 1000; } destroySchedulerFuture = destroyScheduler.scheduleAtFixedRate(destroyTask, period, period, - TimeUnit.MILLISECONDS); + TimeUnit.MILLISECONDS); initedLatch.countDown(); return; } @@ -817,7 +842,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; @@ -828,7 +853,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat /** * load filters from SPI ServiceLoader - * + * * @see ServiceLoader */ private void initFromSPIServiceLoader() { @@ -873,7 +898,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat /** * 会去重复 - * + * * @param filter */ private void addFilter(Filter filter) { @@ -928,7 +953,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()) { @@ -965,7 +990,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"); } } } @@ -994,7 +1019,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"); } } } @@ -1002,14 +1027,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(); @@ -1027,7 +1052,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)) { @@ -1055,7 +1080,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); @@ -1076,7 +1101,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 { @@ -1157,7 +1182,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat /** * 抛弃连接，不进行回收，而是抛弃 - * + * * @param realConnection */ public void discardConnection(Connection realConnection) { @@ -1203,7 +1228,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()); } } @@ -1235,7 +1260,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); @@ -1249,7 +1274,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat .append(", maxActive " + maxActive)// ; JdbcDataSourceStat sourceStat = this.getDataSourceStat(); - if(sourceStat != null) { + if (sourceStat != null) { List sqlList = sourceStat.getRuningSqlList(); for (int i = 0; i < sqlList.size(); ++i) { if (i != 0) { @@ -1344,9 +1369,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"); } @@ -1543,7 +1568,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; @@ -1633,7 +1658,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 @@ -1654,8 +1679,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); @@ -1926,7 +1951,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat createTaskCount--; if (poolingCount + createTaskCount < notEmptyWaitThreadCount // - && activeCount + poolingCount + createTaskCount < maxActive) { + && activeCount + poolingCount + createTaskCount < maxActive) { emptySignal(); } } @@ -1946,7 +1971,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } private void runInternal() { - for (;;) { + for (; ; ) { // addLast lock.lock(); @@ -2043,8 +2068,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); } @@ -2054,7 +2080,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat long lastDiscardCount = 0; int errorCount = 0; - for (;;) { + for (; ; ) { // addLast try { lock.lockInterruptibly(); @@ -2102,7 +2128,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) { @@ -2126,6 +2152,8 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } catch (InterruptedException interruptEx) { break; } + DruidDataSource.this.createThreadChange(); + break; } } catch (RuntimeException e) { LOG.error("create connection error", e); @@ -2150,11 +2178,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); } @@ -2162,7 +2199,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat public void run() { initedLatch.countDown(); - for (;;) { + for (; ; ) { // 从前面开始删除 try { if (closed) { @@ -2186,6 +2223,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 { @@ -2203,14 +2247,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) { @@ -2236,7 +2280,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()) { @@ -2288,7 +2332,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 "); @@ -2304,7 +2348,9 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat return removeCount; } - /** Instance key */ + /** + * Instance key + */ protected String instanceKey = null; public Reference getReference() throws NamingException { @@ -2938,7 +2984,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; } @@ -2954,7 +3000,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; } @@ -2998,7 +3044,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } int fillCount = 0; - for (;;) { + for (; ; ) { try { lock.lockInterruptibly(); } catch (InterruptedException e) { From 745fe8584ea0b17d318370edb66b67e91142f8fe Mon Sep 17 00:00:00 2001 From: abel Date: Mon, 6 Aug 2018 10:43:54 +0800 Subject: [PATCH 2/3] fix import --- .../src/com/fr/third/alibaba/druid/pool/DruidDataSource.java | 1 - 1 file changed, 1 deletion(-) 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 dc0c3b042..da0f2bed4 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 @@ -35,7 +35,6 @@ 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; From 68085657b2a1158c5e8b21e88ad0a5c814f48eb1 Mon Sep 17 00:00:00 2001 From: abel Date: Mon, 6 Aug 2018 11:17:31 +0800 Subject: [PATCH 3/3] =?UTF-8?q?=E7=BA=BF=E7=A8=8B=E5=88=87=E6=8D=A2?= =?UTF-8?q?=EF=BC=8C=E5=A4=9A=E7=BA=BF=E7=A8=8B=E5=90=8C=E6=AD=A5=E9=97=AE?= =?UTF-8?q?=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../alibaba/druid/pool/DruidDataSource.java | 20 +++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) 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 da0f2bed4..567ccd5c8 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 @@ -614,14 +614,18 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat 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); + synchronized (this) {//线程安全问题，加个双检锁 + 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); + } + } } } }