diff --git a/src/main/java/com/alibaba/druid/pool/DruidConnectionHolder.java b/src/main/java/com/alibaba/druid/pool/DruidConnectionHolder.java index 6a755ceff2..32823227ca 100644 --- a/src/main/java/com/alibaba/druid/pool/DruidConnectionHolder.java +++ b/src/main/java/com/alibaba/druid/pool/DruidConnectionHolder.java @@ -18,10 +18,12 @@ package com.alibaba.druid.pool; import com.alibaba.druid.support.logging.Log; import com.alibaba.druid.support.logging.LogFactory; import com.alibaba.druid.util.JdbcConstants; +import com.alibaba.druid.util.JdbcUtils; import com.alibaba.druid.util.Utils; import javax.sql.ConnectionEventListener; import javax.sql.StatementEventListener; + import java.sql.Connection; import java.sql.SQLException; import java.sql.Statement; @@ -232,7 +234,7 @@ public final class DruidConnectionHolder { for (Object item : statementTrace.toArray()) { Statement stmt = (Statement) item; - stmt.close(); + JdbcUtils.close(stmt); } statementTrace.clear(); diff --git a/src/main/java/com/alibaba/druid/pool/DruidDataSource.java b/src/main/java/com/alibaba/druid/pool/DruidDataSource.java index e140e023af..7994771916 100644 --- a/src/main/java/com/alibaba/druid/pool/DruidDataSource.java +++ b/src/main/java/com/alibaba/druid/pool/DruidDataSource.java @@ -128,7 +128,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat // threads private ScheduledFuture destroySchedulerFuture; private DestroyTask destoryTask; - + private CreateConnectionThread createConnectionThread; private DestroyConnectionThread destroyConnectionThread; private LogStatsThread logStatsThread; @@ -155,7 +155,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat public static ThreadLocal waitNanosLocal = new ThreadLocal(); private boolean logDifferentThread = true; - + public DruidDataSource(){ this(false); } @@ -527,7 +527,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat this.jdbcUrl = this.jdbcUrl.trim(); initFromWrapDriverUrl(); } - + if (this.dbType == null || this.dbType.length() == 0) { this.dbType = JdbcUtils.getDbType(jdbcUrl, null); } @@ -671,17 +671,18 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat protected void createAndStartDestroyThread() { destoryTask = new DestroyTask(); - + if (destroyScheduler != null) { long period = timeBetweenEvictionRunsMillis; if (period <= 0) { period = 1000; } - destroySchedulerFuture = destroyScheduler.scheduleAtFixedRate(destoryTask, period, period, TimeUnit.MILLISECONDS); + destroySchedulerFuture = destroyScheduler.scheduleAtFixedRate(destoryTask, period, period, + TimeUnit.MILLISECONDS); initedLatch.countDown(); return; } - + String threadName = "Druid-ConnectionPool-Destroy-" + System.identityHashCode(this); destroyConnectionThread = new DestroyConnectionThread(threadName); destroyConnectionThread.start(); @@ -694,7 +695,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat createConnectionThread.start(); return; } - + initedLatch.countDown(); } @@ -883,6 +884,10 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } private void initExceptionSorter() { + if (this.exceptionSorter != null) { + return; + } + String realDriverClassName = driver.getClass().getName(); if (realDriverClassName.equals(JdbcConstants.MYSQL_DRIVER)) { this.exceptionSorter = new MySqlExceptionSorter(); @@ -1152,10 +1157,20 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } } } - this.discardConnection(holder.getConnection()); - holder.setDiscard(true); - pooledConnection.disable(t); + boolean requireDiscard = false; + synchronized (pooledConnection) { + if ((!pooledConnection.isClosed()) || !pooledConnection.isDisable()) { + holder.setDiscard(true); + pooledConnection.disable(t); + requireDiscard = true; + } + } + + if (requireDiscard) { + this.discardConnection(holder.getConnection()); + } + LOG.error("discard connection", sqlEx); } @@ -1177,9 +1192,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"); } @@ -1300,7 +1315,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat if (destroyConnectionThread != null) { destroyConnectionThread.interrupt(); } - + if (destroySchedulerFuture != null) { destroySchedulerFuture.cancel(true); } @@ -1479,11 +1494,11 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat return last; } } - + private final void decrementPoolingCount() { poolingCount--; } - + private final void incrementPoolingCount() { poolingCount++; } @@ -1670,7 +1685,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat public long getRemoveAbandonedCount() { return removeAbandonedCount; } - + protected void put(Connection connection) { DruidConnectionHolder holder = null; try { @@ -1692,10 +1707,10 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat notEmpty.signal(); notEmptySignalCount++; - + if (createScheduler != null) { createTaskCount--; - + if (poolingCount + createTaskCount < notEmptyWaitThreadCount // && activeCount + poolingCount + createTaskCount < maxActive) { emptySignal(); @@ -1705,10 +1720,11 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat lock.unlock(); } } - + public class CreateConnectionTask implements Runnable { + private int errorCount = 0; - + @Override public void run() { for (;;) { @@ -1843,7 +1859,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } put(connection); - + errorCount = 0; // reset errorCount } } @@ -1884,7 +1900,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } } - + public class DestroyTask implements Runnable { @Override @@ -1895,7 +1911,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat removeAbandoned(); } } - + } public class LogStatsThread extends Thread { @@ -2556,7 +2572,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat public void setLogDifferentThread(boolean logDifferentThread) { this.logDifferentThread = logDifferentThread; } - + public DruidPooledConnection tryGetConnection() throws SQLException { if (poolingCount == 0) { return null; @@ -2646,7 +2662,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat return true; } } - + public boolean isFull() { lock.lock(); try { @@ -2655,21 +2671,21 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat lock.unlock(); } } - + private void emptySignal() { if (createScheduler == null) { empty.signal(); return; } - + if (createTaskCount >= maxCreateTaskCount) { return; } - + if (activeCount + poolingCount + createTaskCount >= maxActive) { return; } - + createTaskCount++; CreateConnectionTask task = new CreateConnectionTask(); createScheduler.submit(task); diff --git a/src/main/java/com/alibaba/druid/support/logging/Log4jImpl.java b/src/main/java/com/alibaba/druid/support/logging/Log4jImpl.java index 7b63281283..f93c13f882 100644 --- a/src/main/java/com/alibaba/druid/support/logging/Log4jImpl.java +++ b/src/main/java/com/alibaba/druid/support/logging/Log4jImpl.java @@ -40,6 +40,10 @@ public class Log4jImpl implements Log { public Log4jImpl(String loggerName){ log = Logger.getLogger(loggerName); } + + public Logger getLog() { + return log; + } public boolean isDebugEnabled() { return log.isDebugEnabled(); diff --git a/src/test/java/com/alibaba/druid/bvt/pool/AsyncCloseTest.java b/src/test/java/com/alibaba/druid/bvt/pool/AsyncCloseTest.java index c0d0a9b81d..794785c3a8 100644 --- a/src/test/java/com/alibaba/druid/bvt/pool/AsyncCloseTest.java +++ b/src/test/java/com/alibaba/druid/bvt/pool/AsyncCloseTest.java @@ -32,6 +32,10 @@ public class AsyncCloseTest extends TestCase { closeExecutor = Executors.newFixedThreadPool(100); } + + protected void tearDown() throws Exception { + dataSource.close(); + } public void test_0() throws Exception { for (int i = 0; i < 16; ++i) { diff --git a/src/test/java/com/alibaba/druid/bvt/pool/AsyncCloseTest2.java b/src/test/java/com/alibaba/druid/bvt/pool/AsyncCloseTest2.java new file mode 100644 index 0000000000..20a5414522 --- /dev/null +++ b/src/test/java/com/alibaba/druid/bvt/pool/AsyncCloseTest2.java @@ -0,0 +1,147 @@ +package com.alibaba.druid.bvt.pool; + +import java.sql.Connection; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; + +import junit.framework.TestCase; + +import org.apache.log4j.Level; +import org.apache.log4j.Logger; +import org.junit.Assert; + +import com.alibaba.druid.filter.FilterAdapter; +import com.alibaba.druid.filter.FilterChain; +import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.druid.pool.vendor.MockExceptionSorter; +import com.alibaba.druid.proxy.jdbc.StatementProxy; +import com.alibaba.druid.support.logging.Log4jImpl; +import com.alibaba.druid.support.logging.LogFactory; + +public class AsyncCloseTest2 extends TestCase { + + protected DruidDataSource dataSource; + private ExecutorService connExecutor; + private ExecutorService closeExecutor; + + final AtomicInteger errorCount = new AtomicInteger(); + + private Logger log; + private Level oldLevel; + + protected void setUp() throws Exception { + log = ((Log4jImpl) LogFactory.getLog(DruidDataSource.class)).getLog(); + oldLevel = log.getLevel(); + log.setLevel(Level.FATAL); + + dataSource = new DruidDataSource(); + dataSource.setUrl("jdbc:mock:"); + dataSource.setAsyncCloseConnectionEnable(true); + dataSource.setTestOnBorrow(false); + dataSource.setMaxActive(16); + + dataSource.getProxyFilters().add(new FilterAdapter() { + + @Override + public boolean statement_execute(FilterChain chain, StatementProxy statement, String sql) + throws SQLException { + throw new SQLException(); + } + }); + + dataSource.setExceptionSorter(new MyExceptionSorter()); + dataSource.init(); + + connExecutor = Executors.newFixedThreadPool(128); + closeExecutor = Executors.newFixedThreadPool(128); + + } + + protected void tearDown() throws Exception { + dataSource.close(); + log.setLevel(oldLevel); + } + + public void test_0() throws Exception { + for (int i = 0; i < 16; ++i) { + loop(); + System.out.println("loop " + i + " done."); + } + } + + public static class MyExceptionSorter extends MockExceptionSorter { + + @Override + public boolean isExceptionFatal(SQLException e) { + return true; + } + } + + class CloseTask implements Runnable { + + private Connection conn; + private CountDownLatch latch; + + public CloseTask(Connection conn, CountDownLatch latch){ + this.conn = conn; + this.latch = latch; + } + + @Override + public void run() { + try { + conn.close(); + } catch (SQLException e) { + errorCount.incrementAndGet(); + } finally { + latch.countDown(); + } + } + + } + + protected void loop() throws InterruptedException { + dataSource.shrink(); + Assert.assertEquals(0, dataSource.getActiveCount()); + Assert.assertEquals(0, dataSource.getPoolingCount()); + + final int COUNT = 1024 * 128; + final CountDownLatch closeLatch = new CountDownLatch(COUNT * 2); + + Runnable connTask = new Runnable() { + + @Override + public void run() { + try { + Connection conn = dataSource.getConnection(); + Statement stmt = conn.createStatement(); + + CloseTask closeTask = new CloseTask(conn, closeLatch); + + closeExecutor.submit(closeTask); + closeExecutor.submit(closeTask); // dup close + + try { + stmt.execute("select 1"); + } finally { + stmt.close(); + conn.close(); + } + } catch (SQLException e) { + errorCount.incrementAndGet(); + } + } + }; + + for (int i = 0; i < COUNT; ++i) { + connExecutor.submit(connTask); + } + + closeLatch.await(); + Assert.assertEquals(0, dataSource.getActiveCount()); + } +}