Browse Source

druidDatasource断线重连机制有点问题

release/9.0
abel.chen 6 years ago
parent
commit
16d4dbe6f2
  1. 325
      fine-druid/src/com/fr/third/alibaba/druid/pool/DruidDataSource.java

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

@ -15,6 +15,43 @@
*/
package com.fr.third.alibaba.druid.pool;
import java.io.Closeable;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.ConcurrentModificationException;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.ServiceLoader;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import javax.management.JMException;
import javax.management.MBeanRegistration;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import javax.naming.NamingException;
import javax.naming.Reference;
import javax.naming.Referenceable;
import javax.naming.StringRefAddr;
import javax.sql.ConnectionEvent;
import javax.sql.ConnectionEventListener;
import javax.sql.ConnectionPoolDataSource;
import javax.sql.PooledConnection;
import com.fr.third.alibaba.druid.Constants;
import com.fr.third.alibaba.druid.TransactionTimeoutException;
import com.fr.third.alibaba.druid.VERSION;
@ -58,121 +95,84 @@ import com.fr.third.alibaba.druid.util.Utils;
import com.fr.third.alibaba.druid.wall.WallFilter;
import com.fr.third.alibaba.druid.wall.WallProviderStatValue;
import javax.management.JMException;
import javax.management.MBeanRegistration;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import javax.naming.NamingException;
import javax.naming.Reference;
import javax.naming.Referenceable;
import javax.naming.StringRefAddr;
import javax.sql.ConnectionEvent;
import javax.sql.ConnectionEventListener;
import javax.sql.ConnectionPoolDataSource;
import javax.sql.PooledConnection;
import java.io.Closeable;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.ConcurrentModificationException;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.ServiceLoader;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* @author ljw [ljw2083@alibaba-inc.com]
* @author wenshao [szujobs@hotmail.com]
*/
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 CreateConnectionThread createConnectionThread;
private DestroyConnectionThread destroyConnectionThread;
private LogStatsThread logStatsThread;
private int createTaskCount;
private ScheduledFuture<?> destroySchedulerFuture;
private DestroyTask destroyTask;
private final CountDownLatch initedLatch = new CountDownLatch(2);
private CreateConnectionThread createConnectionThread;
private DestroyConnectionThread destroyConnectionThread;
private LogStatsThread logStatsThread;
private int createTaskCount;
private volatile boolean enable = true;
private CountDownLatch initedLatch = new CountDownLatch(2);
private boolean resetStatEnable = true;
private final AtomicLong resetCount = new AtomicLong();
private volatile boolean enable = true;
private String initStackTrace;
private boolean resetStatEnable = true;
private final AtomicLong resetCount = new AtomicLong();
private volatile boolean closed = false;
private long closeTimeMillis = -1L;
private String initStackTrace;
protected JdbcDataSourceStat dataSourceStat;
private volatile boolean closed = false;
private long closeTimeMillis = -1L;
private boolean useGlobalDataSourceStat = false;
protected JdbcDataSourceStat dataSourceStat;
private boolean mbeanRegistered = false;
private boolean useGlobalDataSourceStat = false;
public static ThreadLocal<Long> waitNanosLocal = new ThreadLocal<Long>();
private boolean mbeanRegistered = false;
private boolean logDifferentThread = true;
public static ThreadLocal<Long> waitNanosLocal = new ThreadLocal<Long>();
private volatile boolean keepAlive = false;
private boolean logDifferentThread = true;
private static final boolean logActiveCount = "true".equals(System.getProperty("frDruidLogActiveCount"));
private volatile boolean keepAlive = false;
public DruidDataSource(){
public DruidDataSource() {
this(false);
}
public DruidDataSource(boolean fairLock){
public DruidDataSource(boolean fairLock) {
super(fairLock);
configFromPropety(System.getProperties());
@ -603,6 +603,30 @@ 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()) {
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;
@ -646,7 +670,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;
@ -806,7 +830,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
period = 1000;
}
destroySchedulerFuture = destroyScheduler.scheduleAtFixedRate(destroyTask, period, period,
TimeUnit.MILLISECONDS);
TimeUnit.MILLISECONDS);
initedLatch.countDown();
return;
}
@ -818,7 +842,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;
@ -929,7 +953,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()) {
@ -966,7 +990,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");
}
}
}
@ -995,7 +1019,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");
}
}
}
@ -1003,14 +1027,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();
@ -1028,7 +1052,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)) {
@ -1056,7 +1080,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);
@ -1077,7 +1101,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 {
@ -1167,7 +1191,6 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
lock.lock();
try {
activeCount--;
logActiveCount(false);
discardCount++;
if (activeCount <= minIdle) {
@ -1205,7 +1228,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());
}
}
@ -1219,7 +1242,6 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
if (holder != null) {
activeCount++;
logActiveCount(true);
if (activeCount > activePeak) {
activePeak = activeCount;
activePeakTime = System.currentTimeMillis();
@ -1238,7 +1260,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);
@ -1251,19 +1273,21 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
.append(", active " + activeCount)//
.append(", maxActive " + maxActive)//
;
List<JdbcSqlStatValue> sqlList = this.getDataSourceStat().getRuningSqlList();
for (int i = 0; i < sqlList.size(); ++i) {
if (i != 0) {
buf.append('\n');
} else {
buf.append(", ");
JdbcDataSourceStat sourceStat = this.getDataSourceStat();
if (sourceStat != null) {
List<JdbcSqlStatValue> sqlList = sourceStat.getRuningSqlList();
for (int i = 0; i < sqlList.size(); ++i) {
if (i != 0) {
buf.append('\n');
} else {
buf.append(", ");
}
JdbcSqlStatValue sql = sqlList.get(i);
buf.append("runningSqlCount ");
buf.append(sql.getRunningCount());
buf.append(" : ");
buf.append(sql.getSql());
}
JdbcSqlStatValue sql = sqlList.get(i);
buf.append("runningSqlCount ");
buf.append(sql.getRunningCount());
buf.append(" : ");
buf.append(sql.getSql());
}
String errorMessage = buf.toString();
throw new GetConnectionTimeoutException(errorMessage);
@ -1345,9 +1369,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");
}
@ -1409,7 +1433,6 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
lock.lock();
try {
activeCount--;
logActiveCount(false);
closeCount++;
} finally {
lock.unlock();
@ -1428,7 +1451,6 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
lock.lockInterruptibly();
try {
activeCount--;
logActiveCount(false);
closeCount++;
result = putLast(holder, lastActiveTimeMillis);
@ -1546,7 +1568,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;
@ -1636,7 +1658,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
@ -1657,8 +1679,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);
@ -1929,7 +1951,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
createTaskCount--;
if (poolingCount + createTaskCount < notEmptyWaitThreadCount //
&& activeCount + poolingCount + createTaskCount < maxActive) {
&& activeCount + poolingCount + createTaskCount < maxActive) {
emptySignal();
}
}
@ -1949,7 +1971,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
}
private void runInternal() {
for (;;) {
for (; ; ) {
// addLast
lock.lock();
@ -2017,9 +2039,6 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
LOG.error("create connection error", e);
// unknow fatal exception
setFailContinuous(true);
if (breakAfterAcquireFailure) {
break;
}
continue;
} catch (Error e) {
lock.lock();
@ -2049,8 +2068,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);
}
@ -2060,7 +2080,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
long lastDiscardCount = 0;
int errorCount = 0;
for (;;) {
for (; ; ) {
// addLast
try {
lock.lockInterruptibly();
@ -2108,7 +2128,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) {
@ -2132,13 +2152,12 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
} catch (InterruptedException interruptEx) {
break;
}
DruidDataSource.this.createThreadChange();
break;
}
} catch (RuntimeException e) {
LOG.error("create connection error", e);
setFailContinuous(true);
if (breakAfterAcquireFailure) {
break;
}
continue;
} catch (Error e) {
LOG.error("create connection error", e);
@ -2159,11 +2178,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);
}
@ -2171,7 +2199,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
public void run() {
initedLatch.countDown();
for (;;) {
for (; ; ) {
// 从前面开始删除
try {
if (closed) {
@ -2195,6 +2223,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 {
@ -2212,14 +2247,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) {
@ -2245,7 +2280,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()) {
@ -2297,7 +2332,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 ");
@ -2313,7 +2348,9 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
return removeCount;
}
/** Instance key */
/**
* Instance key
*/
protected String instanceKey = null;
public Reference getReference() throws NamingException {
@ -2947,7 +2984,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;
}
@ -2963,7 +3000,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;
}
@ -3007,7 +3044,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
}
int fillCount = 0;
for (;;) {
for (; ; ) {
try {
lock.lockInterruptibly();
} catch (InterruptedException e) {
@ -3129,10 +3166,4 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
public boolean isClosed() {
return this.closed;
}
private void logActiveCount(boolean isIncrease){
if (logActiveCount){
LOG.debug("activeCount " + (isIncrease ? "increase" : "decrease") + "current count is : " + activeCount, new Throwable());
}
}
}

Loading…
Cancel
Save