Browse Source

Merge pull request #84 in CORE/base-third from ~ABEL.CHEN/base-third:release/10.0 to release/10.0

* commit '68085657b2a1158c5e8b21e88ad0a5c814f48eb1':
  线程切换,多线程同步问题
  fix import
  druid断线重连机制修复
final/10.0.3
abel.chen 6 years ago
parent
commit
298a36ade7
  1. 221
      fine-druid/src/com/fr/third/alibaba/druid/pool/DruidDataSource.java

221
fine-druid/src/com/fr/third/alibaba/druid/pool/DruidDataSource.java

@ -100,78 +100,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<Long> waitNanosLocal = new ThreadLocal<Long>();
public static ThreadLocal<Long> waitNanosLocal = new ThreadLocal<Long>();
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 +602,34 @@ 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()) {
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);
}
}
}
}
}
public void init() throws SQLException {
if (inited) {
return;
@ -645,7 +673,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 +833,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 +845,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 +856,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
/**
* load filters from SPI ServiceLoader
*
*
* @see ServiceLoader
*/
private void initFromSPIServiceLoader() {
@ -873,7 +901,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
/**
* 会去重复
*
*
* @param filter
*/
private void addFilter(Filter filter) {
@ -928,7 +956,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 +993,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 +1022,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 +1030,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 +1055,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 +1083,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 +1104,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 +1185,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
/**
* 抛弃连接不进行回收而是抛弃
*
*
* @param realConnection
*/
public void discardConnection(Connection realConnection) {
@ -1203,7 +1231,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 +1263,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 +1277,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
.append(", maxActive " + maxActive)//
;
JdbcDataSourceStat sourceStat = this.getDataSourceStat();
if(sourceStat != null) {
if (sourceStat != null) {
List<JdbcSqlStatValue> sqlList = sourceStat.getRuningSqlList();
for (int i = 0; i < sqlList.size(); ++i) {
if (i != 0) {
@ -1344,9 +1372,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 +1571,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 +1661,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 +1682,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 +1954,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
createTaskCount--;
if (poolingCount + createTaskCount < notEmptyWaitThreadCount //
&& activeCount + poolingCount + createTaskCount < maxActive) {
&& activeCount + poolingCount + createTaskCount < maxActive) {
emptySignal();
}
}
@ -1946,7 +1974,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
}
private void runInternal() {
for (;;) {
for (; ; ) {
// addLast
lock.lock();
@ -2043,8 +2071,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 +2083,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
long lastDiscardCount = 0;
int errorCount = 0;
for (;;) {
for (; ; ) {
// addLast
try {
lock.lockInterruptibly();
@ -2102,7 +2131,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 +2155,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 +2181,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 +2202,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
public void run() {
initedLatch.countDown();
for (;;) {
for (; ; ) {
// 从前面开始删除
try {
if (closed) {
@ -2186,6 +2226,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 +2250,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 +2283,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
try {
Iterator<DruidPooledConnection> iter = activeConnections.keySet().iterator();
for (; iter.hasNext();) {
for (; iter.hasNext(); ) {
DruidPooledConnection pooledConnection = iter.next();
if (pooledConnection.isRunning()) {
@ -2288,7 +2335,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 +2351,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 +2987,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 +3003,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 +3047,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
}
int fillCount = 0;
for (;;) {
for (; ; ) {
try {
lock.lockInterruptibly();
} catch (InterruptedException e) {

Loading…
Cancel
Save