From 3fb484aa8807b2477cf9a3b6ddc4fbbaf13f3c4c Mon Sep 17 00:00:00 2001 From: wenshao Date: Mon, 20 Nov 2017 22:55:23 +0800 Subject: [PATCH] support create direct. --- .../druid/pool/DruidAbstractDataSource.java | 14 +- .../alibaba/druid/pool/DruidDataSource.java | 175 ++++++++++++------ .../CreateSchedulerTest_directCreate.java | 112 +++++++++++ 3 files changed, 236 insertions(+), 65 deletions(-) create mode 100644 src/test/java/com/alibaba/druid/bvt/pool/CreateSchedulerTest_directCreate.java diff --git a/src/main/java/com/alibaba/druid/pool/DruidAbstractDataSource.java b/src/main/java/com/alibaba/druid/pool/DruidAbstractDataSource.java index 84a04fc64b..6f72fe10e0 100644 --- a/src/main/java/com/alibaba/druid/pool/DruidAbstractDataSource.java +++ b/src/main/java/com/alibaba/druid/pool/DruidAbstractDataSource.java @@ -37,9 +37,7 @@ import java.util.Properties; import java.util.Set; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.*; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; import java.util.logging.Logger; @@ -235,9 +233,10 @@ public abstract class DruidAbstractDataSource extends WrapperAdapter implements protected ReentrantLock activeConnectionLock = new ReentrantLock(); - protected AtomicInteger creatingCount = new AtomicInteger(); + protected volatile int creatingCount = 0; protected AtomicLong createCount = new AtomicLong(); protected AtomicLong destroyCount = new AtomicLong(); + protected volatile int directCreateCount = 0; private Boolean useUnfairLock = null; @@ -258,6 +257,9 @@ public abstract class DruidAbstractDataSource extends WrapperAdapter implements protected volatile boolean onFatalError = false; protected volatile int onFatalErrorMaxActive = 0; + protected static AtomicIntegerFieldUpdater directCreateCountUpdater = AtomicIntegerFieldUpdater.newUpdater(DruidAbstractDataSource.class, "directCreateCount"); + protected static AtomicIntegerFieldUpdater createCountUpdater = AtomicIntegerFieldUpdater.newUpdater(DruidAbstractDataSource.class, "creatingCount"); + public DruidAbstractDataSource(boolean lockFair){ lock = new ReentrantLock(lockFair); @@ -1562,7 +1564,7 @@ public abstract class DruidAbstractDataSource extends WrapperAdapter implements ? new HashMap() : null; - creatingCount.incrementAndGet(); + createCountUpdater.incrementAndGet(this); try { conn = createPhysicalConnection(url, physicalConnectProperties); connectedNanos = System.nanoTime(); @@ -1594,7 +1596,7 @@ public abstract class DruidAbstractDataSource extends WrapperAdapter implements } finally { long nano = System.nanoTime() - connectStartNanos; createTimespan += nano; - creatingCount.decrementAndGet(); + createCountUpdater.decrementAndGet(this); } return new PhysicalConnectionInfo(conn, connectStartNanos, connectedNanos, initedNanos, validatedNanos, variables, globalVariables); diff --git a/src/main/java/com/alibaba/druid/pool/DruidDataSource.java b/src/main/java/com/alibaba/druid/pool/DruidDataSource.java index 30fe4a033f..4077f4edaa 100644 --- a/src/main/java/com/alibaba/druid/pool/DruidDataSource.java +++ b/src/main/java/com/alibaba/druid/pool/DruidDataSource.java @@ -24,11 +24,10 @@ import java.sql.Connection; import java.sql.SQLException; import java.sql.Statement; import java.util.*; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.Future; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicLongFieldUpdater; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -93,10 +92,11 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat private final static Log LOG = LogFactory.getLog(DruidDataSource.class); private static final long serialVersionUID = 1L; // stats - private final AtomicLong recycleErrorCount = new AtomicLong(); + private volatile long recycleErrorCount = 0L; private long connectCount = 0L; private long closeCount = 0L; - private final AtomicLong connectErrorCount = new AtomicLong(); + private volatile long connectErrorCount = 0L; +// private final AtomicLong connectErrorCount = new AtomicLong(); private long recycleCount = 0L; private long removeAbandonedCount = 0L; private long notEmptyWaitCount = 0L; @@ -152,6 +152,11 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat private boolean asyncInit = false; protected boolean killWhenSocketReadTimeout = false; + protected static final AtomicLongFieldUpdater recycleErrorCountUpdater + = AtomicLongFieldUpdater.newUpdater(DruidDataSource.class, "recycleErrorCount"); + protected static final AtomicLongFieldUpdater connectErrorCountUpdater + = AtomicLongFieldUpdater.newUpdater(DruidDataSource.class, "connectErrorCount"); + public DruidDataSource(){ this(false); } @@ -548,7 +553,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat lock.unlock(); } - connectErrorCount.set(0); + connectErrorCountUpdater.set(this, 0); errorCount.set(0); commitCount.set(0); rollbackCount.set(0); @@ -558,7 +563,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat preparedStatementCount.set(0); transactionHistogram.reset(); cachedPreparedStatementDeleteCount.set(0); - recycleErrorCount.set(0); + recycleErrorCountUpdater.set(this, 0); resetCount.incrementAndGet(); } @@ -1341,12 +1346,12 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat private DruidPooledConnection getConnectionInternal(long maxWait) throws SQLException { if (closed) { - connectErrorCount.incrementAndGet(); + connectErrorCountUpdater.incrementAndGet(this); throw new DataSourceClosedException("dataSource already closed at " + new Date(closeTimeMillis)); } if (!enable) { - connectErrorCount.incrementAndGet(); + connectErrorCountUpdater.incrementAndGet(this); throw new DataSourceDisableException(); } @@ -1354,52 +1359,104 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat final int maxWaitThreadCount = this.maxWaitThreadCount; DruidConnectionHolder holder; - try { - lock.lockInterruptibly(); - } catch (InterruptedException e) { - connectErrorCount.incrementAndGet(); - throw new SQLException("interrupt", e); - } - try { - if (maxWaitThreadCount > 0 - && notEmptyWaitThreadCount >= maxWaitThreadCount) - { - connectErrorCount.incrementAndGet(); - throw new SQLException("maxWaitThreadCount " + maxWaitThreadCount + ", current wait Thread count " - + lock.getQueueLength()); - } + for (boolean createDirect = false;;) { + if (createDirect) { + if (createCountUpdater.compareAndSet(this, 0, 1)) { + PhysicalConnectionInfo pyConnInfo = DruidDataSource.this.createPhysicalConnection(); + holder = new DruidConnectionHolder(this, pyConnInfo); + holder.lastActiveTimeMillis = System.currentTimeMillis(); - if (onFatalError - && onFatalErrorMaxActive > 0 - && activeCount >= onFatalErrorMaxActive) { - connectErrorCount.incrementAndGet(); - throw new SQLException("onFatalError, activeCount " + activeCount + ", onFatalErrorMaxActive " + onFatalErrorMaxActive); - } + createCountUpdater.decrementAndGet(this); + directCreateCountUpdater.incrementAndGet(this); - connectCount++; + if (LOG.isDebugEnabled()) { + LOG.debug("conn-direct_create "); + } - if (maxWait > 0) { - holder = pollLast(nanos); - } else { - holder = takeLast(); - } + boolean discard = false; + lock.lock(); + try { + if (activeCount < maxActive) { + activeCount++; + if (activeCount > activePeak) { + activePeak = activeCount; + activePeakTime = System.currentTimeMillis(); + } + break; + } else { + discard = true; + } + } finally { + lock.unlock(); + } - if (holder != null) { - activeCount++; - if (activeCount > activePeak) { - activePeak = activeCount; - activePeakTime = System.currentTimeMillis(); + if (discard) { + JdbcUtils.close(pyConnInfo.getPhysicalConnection()); + } } } - } catch (InterruptedException e) { - connectErrorCount.incrementAndGet(); - throw new SQLException(e.getMessage(), e); - } catch (SQLException e) { - connectErrorCount.incrementAndGet(); - throw e; - } finally { - lock.unlock(); + + try { + lock.lockInterruptibly(); + } catch (InterruptedException e) { + connectErrorCountUpdater.incrementAndGet(this); + throw new SQLException("interrupt", e); + } + + try { + if (maxWaitThreadCount > 0 + && notEmptyWaitThreadCount >= maxWaitThreadCount) { + connectErrorCountUpdater.incrementAndGet(this); + throw new SQLException("maxWaitThreadCount " + maxWaitThreadCount + ", current wait Thread count " + + lock.getQueueLength()); + } + + if (onFatalError + && onFatalErrorMaxActive > 0 + && activeCount >= onFatalErrorMaxActive) { + connectErrorCountUpdater.incrementAndGet(this); + throw new SQLException("onFatalError, activeCount " + activeCount + ", onFatalErrorMaxActive " + onFatalErrorMaxActive); + } + + connectCount++; + + if (createScheduler != null + && poolingCount == 0 + && activeCount < maxActive + && createCountUpdater.get(this) == 0 + && createScheduler instanceof ScheduledThreadPoolExecutor) { + ScheduledThreadPoolExecutor executor = (ScheduledThreadPoolExecutor) createScheduler; + if (executor.getQueue().size() > 0) { + createDirect = true; + continue; + } + } + + if (maxWait > 0) { + holder = pollLast(nanos); + } else { + holder = takeLast(); + } + + if (holder != null) { + activeCount++; + if (activeCount > activePeak) { + activePeak = activeCount; + activePeakTime = System.currentTimeMillis(); + } + } + } catch (InterruptedException e) { + connectErrorCountUpdater.incrementAndGet(this); + throw new SQLException(e.getMessage(), e); + } catch (SQLException e) { + connectErrorCountUpdater.incrementAndGet(this); + throw e; + } finally { + lock.unlock(); + } + + break; } if (holder == null) { @@ -1410,7 +1467,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat .append(waitNanos / (1000 * 1000))// .append(", active ").append(activeCount)// .append(", maxActive ").append(maxActive)// - .append(", creating ").append(creatingCount.get())// + .append(", creating ").append(creatingCount)// ; List sqlList = this.getDataSourceStat().getRuningSqlList(); @@ -1629,12 +1686,12 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } LOG.error("recyle error", e); - recycleErrorCount.incrementAndGet(); + recycleErrorCountUpdater.incrementAndGet(this); } } public long getRecycleErrorCount() { - return recycleErrorCount.get(); + return recycleErrorCountUpdater.get(this); } public void clearStatementCache() throws SQLException { @@ -1804,7 +1861,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat notEmptyWaitCount++; if (!enable) { - connectErrorCount.incrementAndGet(); + connectErrorCountUpdater.incrementAndGet(this); throw new DataSourceDisableException(); } } @@ -1851,7 +1908,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat notEmptyWaitNanos += (startEstimate - estimate); if (!enable) { - connectErrorCount.incrementAndGet(); + connectErrorCountUpdater.incrementAndGet(this); throw new DataSourceDisableException(); } } catch (InterruptedException ie) { @@ -1936,7 +1993,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } public long getConnectErrorCount() { - return connectErrorCount.get(); + return connectErrorCountUpdater.get(this); } @Override @@ -2054,7 +2111,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } value.setDefaultTransactionIsolation(this.getDefaultTransactionIsolation()); - value.setLogicConnectErrorCount(connectErrorCount.getAndSet(0)); + value.setLogicConnectErrorCount(connectErrorCountUpdater.getAndSet(this, 0)); value.setPhysicalConnectCount(createCount.getAndSet(0)); value.setPhysicalCloseCount(destroyCount.getAndSet(0)); @@ -3260,7 +3317,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat try { lock.lockInterruptibly(); } catch (InterruptedException e) { - connectErrorCount.incrementAndGet(); + connectErrorCountUpdater.incrementAndGet(this); throw new SQLException("interrupt", e); } @@ -3278,14 +3335,14 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat holder = new DruidConnectionHolder(this, pyConnInfo); } catch (SQLException e) { LOG.error("fill connection error, url: " + this.jdbcUrl, e); - connectErrorCount.incrementAndGet(); + connectErrorCountUpdater.incrementAndGet(this); throw e; } try { lock.lockInterruptibly(); } catch (InterruptedException e) { - connectErrorCount.incrementAndGet(); + connectErrorCountUpdater.incrementAndGet(this); throw new SQLException("interrupt", e); } diff --git a/src/test/java/com/alibaba/druid/bvt/pool/CreateSchedulerTest_directCreate.java b/src/test/java/com/alibaba/druid/bvt/pool/CreateSchedulerTest_directCreate.java new file mode 100644 index 0000000000..e35ae11330 --- /dev/null +++ b/src/test/java/com/alibaba/druid/bvt/pool/CreateSchedulerTest_directCreate.java @@ -0,0 +1,112 @@ +package com.alibaba.druid.bvt.pool; + +import com.alibaba.druid.PoolTestCase; +import com.alibaba.druid.mock.MockConnection; +import com.alibaba.druid.mock.MockDriver; +import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.druid.util.JdbcUtils; +import org.junit.Assert; + +import java.sql.Connection; +import java.sql.SQLException; +import java.util.Properties; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; + +public class CreateSchedulerTest_directCreate extends PoolTestCase { + + private DruidDataSource[] dataSources; + private ScheduledExecutorService createScheduler; + private ScheduledExecutorService destroyScheduler; + + protected void setUp() throws Exception { + super.setUp(); + + createScheduler = Executors.newScheduledThreadPool(1); + destroyScheduler = Executors.newScheduledThreadPool(1); + + dataSources = new DruidDataSource[8]; + for (int i = 0; i < dataSources.length; ++i) { + DruidDataSource dataSource = new DruidDataSource(); + dataSource.setUrl("jdbc:mock:xxx"); + dataSource.setDriver(new SlowDriver()); + dataSource.setCreateScheduler(createScheduler); + dataSource.setDestroyScheduler(destroyScheduler); + dataSource.setMinEvictableIdleTimeMillis(10); + dataSource.setTimeBetweenEvictionRunsMillis(10); + dataSource.setFilters("log4j"); + dataSource.setName("ds_" + i); + + dataSources[i] = dataSource; + } + } + + protected void tearDown() throws Exception { + for (int i = 0; i < dataSources.length; ++i) { + JdbcUtils.close(dataSources[i]); + } + + createScheduler.shutdown(); + + super.tearDown(); + } + + public void test_connectAndClose() throws Exception { + Thread[] threads = new Thread[32]; + final CountDownLatch latch = new CountDownLatch(threads.length); + for (int i = 0; i < threads.length; ++i) { + final int threadId = i; + int dataSourceIndex = i % dataSources.length; + final DruidDataSource dataSource = dataSources[dataSourceIndex]; + Thread thread = new Thread(new Task(dataSource, latch), "thread-" + threadId); + threads[i] = thread; + } + + for (int i = 0; i < threads.length; ++i) { + threads[i].start(); + Thread.sleep(10); + } + + latch.await(); + } + + public static class SlowDriver extends MockDriver { + public MockConnection createMockConnection(MockDriver driver, String url, Properties connectProperties) { + try { + Thread.sleep(1000 * 1); + } catch (InterruptedException e) { + e.printStackTrace(); + } + return super.createMockConnection(driver, url, connectProperties); + } + } + + public static class Task implements Runnable { + private final DruidDataSource dataSource; + private final CountDownLatch latch; + + public Task(DruidDataSource dataSource, CountDownLatch latch) { + this.dataSource = dataSource; + this.latch = latch; + } + + @Override + public void run() { + try { + for (int i = 0; i < 4; ++i) { + Connection conn = null; + try { + conn = dataSource.getConnection(); + } catch (SQLException e) { + e.printStackTrace(); + } finally { + JdbcUtils.close(conn); + } + } + } finally { + latch.countDown(); + } + } + } +}