bug fixed for create/destroy scheduler & add testcase

This commit is contained in:
wenshao
2014-08-22 11:21:44 +08:00
parent d619c0e195
commit 4ade284fee
4 changed files with 204 additions and 10 deletions
@@ -240,6 +240,7 @@ public abstract class DruidAbstractDataSource extends WrapperAdapter implements
protected DruidDataSourceStatLogger statLogger = new DruidDataSourceStatLoggerImpl();
private boolean asyncCloseConnectionEnable = false;
protected int maxCreateTaskCount = 2;
protected ScheduledExecutorService destroyScheduler;
protected ScheduledExecutorService createScheduler;
@@ -1727,4 +1728,18 @@ public abstract class DruidAbstractDataSource extends WrapperAdapter implements
public boolean isInited() {
return this.inited;
}
public int getMaxCreateTaskCount() {
return maxCreateTaskCount;
}
public void setMaxCreateTaskCount(int maxCreateTaskCount) {
if (maxCreateTaskCount < 1) {
throw new IllegalArgumentException();
}
this.maxCreateTaskCount = maxCreateTaskCount;
}
}
@@ -617,7 +617,8 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
for (int i = 0, size = getInitialSize(); i < size; ++i) {
Connection conn = createPhysicalConnection();
DruidConnectionHolder holder = new DruidConnectionHolder(this, conn);
connections[poolingCount++] = holder;
connections[poolingCount] = holder;
incrementPoolingCount();
}
if (poolingCount > 0) {
@@ -1175,7 +1176,10 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
return;
}
if (logDifferentThread && pooledConnection.getOwnerThread() != Thread.currentThread()) {
if (logDifferentThread //
&& (!isAsyncCloseConnectionEnable()) //
&& pooledConnection.getOwnerThread() != Thread.currentThread()//
) {
LOG.warn("get/close not same thread");
}
@@ -1376,7 +1380,8 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
void putLast(DruidConnectionHolder e, long lastActiveTimeMillis) {
e.setLastActiveTimeMillis(lastActiveTimeMillis);
connections[poolingCount++] = e;
connections[poolingCount] = e;
incrementPoolingCount();
if (poolingCount > poolingPeak) {
poolingPeak = poolingCount;
@@ -1413,7 +1418,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
throw ie;
}
poolingCount--;
decrementPoolingCount();
DruidConnectionHolder last = connections[poolingCount];
connections[poolingCount] = null;
@@ -1467,13 +1472,21 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
}
}
poolingCount--;
decrementPoolingCount();
DruidConnectionHolder last = connections[poolingCount];
connections[poolingCount] = null;
return last;
}
}
private final void decrementPoolingCount() {
poolingCount--;
}
private final void incrementPoolingCount() {
poolingCount++;
}
@Override
public Connection getConnection(String username, String password) throws SQLException {
@@ -1669,7 +1682,8 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
lock.lock();
try {
connections[poolingCount++] = holder;
connections[poolingCount] = holder;
incrementPoolingCount();
if (poolingCount > poolingPeak) {
poolingPeak = poolingCount;
@@ -1681,6 +1695,11 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
if (createScheduler != null) {
createTaskCount--;
if (poolingCount + createTaskCount < notEmptyWaitThreadCount //
&& activeCount + poolingCount + createTaskCount < maxActive) {
emptySignal();
}
}
} finally {
lock.unlock();
@@ -1727,7 +1746,8 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
if (errorCount > connectionErrorRetryAttempts && timeBetweenConnectErrorMillis > 0) {
if (breakAfterAcquireFailure) {
break;
createTaskCount--;
return;
}
createScheduler.schedule(this, timeBetweenConnectErrorMillis, TimeUnit.MILLISECONDS);
@@ -1746,6 +1766,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
}
put(connection);
break;
}
}
}
@@ -2641,13 +2662,16 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
return;
}
if (createTaskCount >= 2) {
if (createTaskCount >= maxCreateTaskCount) {
return;
}
CreateConnectionTask task = new CreateConnectionTask();
createScheduler.submit(task);
if (activeCount + poolingCount + createTaskCount >= maxActive) {
return;
}
createTaskCount++;
CreateConnectionTask task = new CreateConnectionTask();
createScheduler.submit(task);
}
}
@@ -0,0 +1,103 @@
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 Large10ConcurrentTest 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 = new Connection[dataSources.length * 8];
final CountDownLatch connLatch = new CountDownLatch(connections.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[index] = dataSource.getConnection();
} catch (SQLException e) {
connErrorCount.incrementAndGet();
e.printStackTrace();
} finally {
connLatch.countDown();
}
}
};
executor.execute(task);
}
}
connLatch.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());
}
final CountDownLatch closeLatch = new CountDownLatch(connections.length);
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[index]);
closeLatch.countDown();
}
};
executor.execute(task);
}
}
closeLatch.await();
}
}
@@ -0,0 +1,52 @@
package com.alibaba.druid.bvt.pool;
import java.sql.Connection;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import junit.framework.TestCase;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.util.JdbcUtils;
public class Large10KTest extends TestCase {
private DruidDataSource[] dataSources = new DruidDataSource[10000];
private ScheduledExecutorService scheduler;
protected void setUp() throws Exception {
scheduler = Executors.newScheduledThreadPool(10);
for (int i = 0; i < dataSources.length; ++i) {
DruidDataSource dataSource = new DruidDataSource();
dataSource.setUrl("jdbc:mock:xxx");
dataSource.setCreateScheduler(scheduler);
dataSource.setDestroyScheduler(scheduler);
dataSource.setTestOnBorrow(false);
dataSource.setTestWhileIdle(false);
dataSources[i] = dataSource;
}
}
protected void tearDown() throws Exception {
for (int i = 0; i < dataSources.length; ++i) {
JdbcUtils.close(dataSources[i]);
}
scheduler.shutdown();
}
public void test_large() throws Exception {
Connection[] connections = new Connection[dataSources.length * 8];
for (int i = 0; i < dataSources.length; ++i) {
for (int j = 0; j < 8; ++j) {
connections[i * 8 + j] = dataSources[i].getConnection();
}
}
for (int i = 0; i < dataSources.length; ++i) {
for (int j = 0; j < 8; ++j) {
connections[i * 8 + j].close();
}
}
}
}