mirror of
https://github.com/alibaba/druid.git
synced 2023-03-04 19:17:16 +00:00
Merge remote-tracking branch 'upstream/master'
This commit is contained in:
@@ -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();
|
||||
|
||||
|
||||
@@ -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<Long> waitNanosLocal = new ThreadLocal<Long>();
|
||||
|
||||
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,21 @@ 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());
|
||||
holder.setDiscard(true);
|
||||
}
|
||||
|
||||
LOG.error("discard connection", sqlEx);
|
||||
}
|
||||
|
||||
@@ -1177,9 +1193,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");
|
||||
}
|
||||
|
||||
@@ -1210,7 +1226,18 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
|
||||
}
|
||||
|
||||
// reset holder, restore default settings, clear warnings
|
||||
holder.reset();
|
||||
boolean isSameThread = pooledConnection.getOwnerThread() == Thread.currentThread();
|
||||
if (!isSameThread) {
|
||||
synchronized (pooledConnection) {
|
||||
holder.reset();
|
||||
}
|
||||
} else {
|
||||
holder.reset();
|
||||
}
|
||||
|
||||
if (holder.isDiscard()) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (testOnReturn) {
|
||||
boolean validate = testConnectionInternal(physicalConnection);
|
||||
@@ -1300,7 +1327,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
|
||||
if (destroyConnectionThread != null) {
|
||||
destroyConnectionThread.interrupt();
|
||||
}
|
||||
|
||||
|
||||
if (destroySchedulerFuture != null) {
|
||||
destroySchedulerFuture.cancel(true);
|
||||
}
|
||||
@@ -1479,11 +1506,11 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
|
||||
return last;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private final void decrementPoolingCount() {
|
||||
poolingCount--;
|
||||
}
|
||||
|
||||
|
||||
private final void incrementPoolingCount() {
|
||||
poolingCount++;
|
||||
}
|
||||
@@ -1670,7 +1697,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
|
||||
public long getRemoveAbandonedCount() {
|
||||
return removeAbandonedCount;
|
||||
}
|
||||
|
||||
|
||||
protected void put(Connection connection) {
|
||||
DruidConnectionHolder holder = null;
|
||||
try {
|
||||
@@ -1692,10 +1719,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 +1732,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 +1871,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
|
||||
}
|
||||
|
||||
put(connection);
|
||||
|
||||
|
||||
errorCount = 0; // reset errorCount
|
||||
}
|
||||
}
|
||||
@@ -1884,7 +1912,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
public class DestroyTask implements Runnable {
|
||||
|
||||
@Override
|
||||
@@ -1895,7 +1923,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
|
||||
removeAbandoned();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
public class LogStatsThread extends Thread {
|
||||
@@ -2556,7 +2584,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 +2674,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public boolean isFull() {
|
||||
lock.lock();
|
||||
try {
|
||||
@@ -2655,21 +2683,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);
|
||||
|
||||
@@ -217,7 +217,14 @@ public class DruidPooledConnection extends PoolableWrapper implements javax.sql.
|
||||
return;
|
||||
}
|
||||
|
||||
if (holder.getDataSource().isAsyncCloseConnectionEnable()) {
|
||||
DruidAbstractDataSource dataSource = holder.getDataSource();
|
||||
boolean isSameThread = this.getOwnerThread() == Thread.currentThread();
|
||||
|
||||
if (!isSameThread) {
|
||||
dataSource.setAsyncCloseConnectionEnable(true);
|
||||
}
|
||||
|
||||
if (dataSource.isAsyncCloseConnectionEnable()) {
|
||||
syncClose();
|
||||
return;
|
||||
}
|
||||
@@ -226,7 +233,7 @@ public class DruidPooledConnection extends PoolableWrapper implements javax.sql.
|
||||
listener.connectionClosed(new ConnectionEvent(this));
|
||||
}
|
||||
|
||||
DruidAbstractDataSource dataSource = holder.getDataSource();
|
||||
|
||||
List<Filter> filters = dataSource.getProxyFilters();
|
||||
if (filters.size() > 0) {
|
||||
FilterChainImpl filterChain = new FilterChainImpl(dataSource);
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
package com.alibaba.druid.bvt.pool;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.sql.SQLException;
|
||||
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.junit.Assert;
|
||||
|
||||
import com.alibaba.druid.pool.DruidDataSource;
|
||||
|
||||
public class AsyncCloseTest extends TestCase {
|
||||
|
||||
protected DruidDataSource dataSource;
|
||||
private ExecutorService connExecutor;
|
||||
private ExecutorService closeExecutor;
|
||||
|
||||
final AtomicInteger errorCount = new AtomicInteger();
|
||||
|
||||
protected void setUp() throws Exception {
|
||||
dataSource = new DruidDataSource();
|
||||
dataSource.setUrl("jdbc:mock:");
|
||||
dataSource.setAsyncCloseConnectionEnable(true);
|
||||
dataSource.setTestOnBorrow(false);
|
||||
dataSource.setMaxActive(16);
|
||||
|
||||
connExecutor = Executors.newFixedThreadPool(100);
|
||||
closeExecutor = Executors.newFixedThreadPool(100);
|
||||
|
||||
}
|
||||
|
||||
protected void tearDown() throws Exception {
|
||||
dataSource.close();
|
||||
}
|
||||
|
||||
public void test_0() throws Exception {
|
||||
for (int i = 0; i < 16; ++i) {
|
||||
loop();
|
||||
System.out.println("loop " + i + " done.");
|
||||
}
|
||||
}
|
||||
|
||||
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();
|
||||
|
||||
CloseTask closeTask = new CloseTask(conn, closeLatch);
|
||||
|
||||
closeExecutor.submit(closeTask);
|
||||
closeExecutor.submit(closeTask); // dup close
|
||||
} catch (SQLException e) {
|
||||
errorCount.incrementAndGet();
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
for (int i = 0; i < COUNT; ++i) {
|
||||
connExecutor.submit(connTask);
|
||||
}
|
||||
|
||||
closeLatch.await();
|
||||
Assert.assertEquals(0, dataSource.getActiveCount());
|
||||
Assert.assertEquals(16, dataSource.getPoolingCount());
|
||||
|
||||
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,117 @@
|
||||
package com.alibaba.druid.bvt.pool;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.sql.SQLException;
|
||||
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.pool.DruidDataSource;
|
||||
import com.alibaba.druid.support.logging.Log4jImpl;
|
||||
import com.alibaba.druid.support.logging.LogFactory;
|
||||
|
||||
public class AsyncCloseTest1 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);
|
||||
|
||||
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.");
|
||||
}
|
||||
}
|
||||
|
||||
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();
|
||||
|
||||
CloseTask closeTask = new CloseTask(conn, closeLatch);
|
||||
|
||||
closeExecutor.submit(closeTask);
|
||||
closeExecutor.submit(closeTask); // dup close
|
||||
} catch (SQLException e) {
|
||||
errorCount.incrementAndGet();
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
for (int i = 0; i < COUNT; ++i) {
|
||||
connExecutor.submit(connTask);
|
||||
}
|
||||
|
||||
closeLatch.await();
|
||||
Assert.assertEquals(0, dataSource.getActiveCount());
|
||||
Assert.assertEquals(16, dataSource.getPoolingCount());
|
||||
|
||||
|
||||
}
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,152 @@
|
||||
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.DruidPooledConnection;
|
||||
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 AsyncCloseTest3 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(2);
|
||||
closeExecutor = Executors.newFixedThreadPool(2);
|
||||
|
||||
}
|
||||
|
||||
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);
|
||||
final CountDownLatch execLatch = new CountDownLatch(COUNT);
|
||||
|
||||
Runnable connTask = new Runnable() {
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
try {
|
||||
DruidPooledConnection 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();
|
||||
} finally {
|
||||
execLatch.countDown();
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
for (int i = 0; i < COUNT; ++i) {
|
||||
connExecutor.submit(connTask);
|
||||
}
|
||||
|
||||
execLatch.await();
|
||||
closeLatch.await();
|
||||
Assert.assertEquals(0, dataSource.getActiveCount());
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -17,7 +17,7 @@ import org.junit.Assert;
|
||||
import com.alibaba.druid.pool.DruidDataSource;
|
||||
import com.alibaba.druid.util.JdbcUtils;
|
||||
|
||||
public class Large10KConcurrentTest extends TestCase {
|
||||
public class Large10ConcurrentTest extends TestCase {
|
||||
|
||||
private DruidDataSource[] dataSources = new DruidDataSource[10000];
|
||||
private ScheduledExecutorService scheduler;
|
||||
@@ -1,142 +0,0 @@
|
||||
package com.alibaba.druid.bvt.pool;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.sql.SQLException;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
|
||||
import junit.framework.TestCase;
|
||||
|
||||
import org.junit.Assert;
|
||||
|
||||
import com.alibaba.druid.pool.DruidDataSource;
|
||||
import com.alibaba.druid.util.JdbcUtils;
|
||||
|
||||
public class Large10KConcurrentTest_2 extends TestCase {
|
||||
|
||||
private DruidDataSource[] dataSources = new DruidDataSource[10000];
|
||||
private ScheduledExecutorService scheduler;
|
||||
|
||||
private ExecutorService executor;
|
||||
|
||||
protected void setUp() throws Exception {
|
||||
executor = Executors.newFixedThreadPool(100);
|
||||
scheduler = Executors.newScheduledThreadPool(10);
|
||||
for (int i = 0; i < dataSources.length; ++i) {
|
||||
DruidDataSource dataSource = new DruidDataSource();
|
||||
dataSource.setMaxActive(8);
|
||||
dataSource.setMinIdle(0);
|
||||
dataSource.setUrl("jdbc:mock:xxx");
|
||||
dataSource.setCreateScheduler(scheduler);
|
||||
dataSource.setDestroyScheduler(scheduler);
|
||||
dataSource.setTestOnBorrow(false);
|
||||
dataSource.setTestWhileIdle(false);
|
||||
dataSource.setAsyncCloseConnectionEnable(true);
|
||||
|
||||
dataSources[i] = dataSource;
|
||||
}
|
||||
}
|
||||
|
||||
protected void tearDown() throws Exception {
|
||||
for (int i = 0; i < dataSources.length; ++i) {
|
||||
JdbcUtils.close(dataSources[i]);
|
||||
}
|
||||
executor.shutdown();
|
||||
scheduler.shutdown();
|
||||
}
|
||||
|
||||
public void test_large() throws Exception {
|
||||
final Connection[] connections_A = new Connection[dataSources.length * 8];
|
||||
final Connection[] connections_B = new Connection[dataSources.length * 8];
|
||||
final CountDownLatch connLatch_A = new CountDownLatch(connections_A.length);
|
||||
final CountDownLatch connLatch_B = new CountDownLatch(connections_A.length);
|
||||
final CountDownLatch closeLatch_A = new CountDownLatch(connections_A.length);
|
||||
final CountDownLatch closeLatch_B = new CountDownLatch(connections_A.length);
|
||||
final AtomicLong connErrorCount = new AtomicLong();
|
||||
|
||||
for (int i = 0; i < dataSources.length; ++i) {
|
||||
for (int j = 0; j < 8; ++j) {
|
||||
final DataSource dataSource = dataSources[i];
|
||||
final int index = i * 8 + j;
|
||||
Runnable task = new Runnable() {
|
||||
|
||||
public void run() {
|
||||
try {
|
||||
connections_A[index] = dataSource.getConnection();
|
||||
} catch (SQLException e) {
|
||||
connErrorCount.incrementAndGet();
|
||||
e.printStackTrace();
|
||||
} finally {
|
||||
connLatch_A.countDown();
|
||||
}
|
||||
}
|
||||
};
|
||||
executor.execute(task);
|
||||
}
|
||||
}
|
||||
connLatch_A.await();
|
||||
|
||||
for (int i = 0; i < dataSources.length; ++i) {
|
||||
Assert.assertEquals(8, dataSources[i].getActiveCount());
|
||||
}
|
||||
|
||||
for (int i = 0; i < dataSources.length; ++i) {
|
||||
Assert.assertEquals(0, dataSources[i].getPoolingCount());
|
||||
}
|
||||
|
||||
for (int i = 0; i < dataSources.length; ++i) {
|
||||
for (int j = 0; j < 8; ++j) {
|
||||
final int index = i * 8 + j;
|
||||
Runnable task = new Runnable() {
|
||||
public void run() {
|
||||
JdbcUtils.close(connections_A[index]);
|
||||
closeLatch_A.countDown();
|
||||
}
|
||||
};
|
||||
executor.execute(task);
|
||||
}
|
||||
}
|
||||
|
||||
for (int i = 0; i < dataSources.length; ++i) {
|
||||
for (int j = 0; j < 8; ++j) {
|
||||
final DataSource dataSource = dataSources[i];
|
||||
final int index = i * 8 + j;
|
||||
Runnable task = new Runnable() {
|
||||
|
||||
public void run() {
|
||||
try {
|
||||
connections_B[index] = dataSource.getConnection();
|
||||
} catch (SQLException e) {
|
||||
connErrorCount.incrementAndGet();
|
||||
e.printStackTrace();
|
||||
} finally {
|
||||
connLatch_B.countDown();
|
||||
}
|
||||
}
|
||||
};
|
||||
executor.execute(task);
|
||||
}
|
||||
}
|
||||
closeLatch_A.await();
|
||||
connLatch_B.await();
|
||||
|
||||
for (int i = 0; i < dataSources.length; ++i) {
|
||||
for (int j = 0; j < 8; ++j) {
|
||||
final int index = i * 8 + j;
|
||||
Runnable task = new Runnable() {
|
||||
public void run() {
|
||||
JdbcUtils.close(connections_B[index]);
|
||||
closeLatch_B.countDown();
|
||||
}
|
||||
};
|
||||
executor.execute(task);
|
||||
}
|
||||
}
|
||||
closeLatch_B.await();
|
||||
}
|
||||
}
|
||||
+2
@@ -63,7 +63,9 @@ public class OracleExceptionSorterTest_closeConn extends TestCase {
|
||||
SQLException exception = new SQLException("xx", "xxx", 28);
|
||||
mockConn.setError(exception);
|
||||
|
||||
Assert.assertEquals(1, dataSource.getActiveCount());
|
||||
conn.close();
|
||||
Assert.assertEquals(0, dataSource.getActiveCount());
|
||||
|
||||
{
|
||||
Connection conn2 = dataSource.getConnection();
|
||||
|
||||
Reference in New Issue
Block a user