opensource
/
fine-third
1
0
Fork 1
Code Issues Pull Requests Releases Wiki Activity
Browse Source

REPORT-71021 恢复创建线程启停机制

release/11.0
rinoux 3 years ago
parent
fb90311894
commit
1d18148c97
2 changed files with 203 additions and 88 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      fine-druid/readme.MD
  2. 207
      fine-druid/src/main/java/com/fr/third/alibaba/druid/pool/DruidDataSource.java

2
fine-druid/readme.MD
Unescape View File

@ -1,4 +1,5 @@
# Alibaba Druid
- FineReport更新时间 `2022-04-27`
- Druid版本 1.2.9
- [github地址](https://github.com/alibaba/druid)
@ -14,3 +15,4 @@
| 1.2.9 | 2022-05-05 | MysqlUtils.getLastPacketReceivedTimeMs根据类加载器区分连接实现等，不在使用全局变量 |
| 1.2.9 | 2022-05-05 | com.fr.third.alibaba.druid.util.Utils.loadClass改为优先从线程类加载器加载类 |
| 1.2.9 | 2022-05-05 | 恢复com.fr.third.alibaba.druid.pool.DruidDataSourceFactory对hibernate配置属性的支持 |
| 1.2.9 | 2022-05-10 | 恢复DruidDataSource中的创建线程启停机制 |

207
fine-druid/src/main/java/com/fr/third/alibaba/druid/pool/DruidDataSource.java
Unescape View File

@ -134,6 +134,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
private volatile Future<?> createSchedulerFuture;
private CreateConnectionThread createConnectionThread;
private PeriodDetectionThread periodDetectionThread;
private DestroyConnectionThread destroyConnectionThread;
private LogStatsThread logStatsThread;
private int createTaskCount;
@ -141,7 +142,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
private volatile long createTaskIdSeed = 1L;
private long[] createTasks;
private final CountDownLatch initedLatch = new CountDownLatch(2);
private CountDownLatch initedLatch = new CountDownLatch(2);
private volatile boolean enable = true;
@ -161,6 +162,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
public static ThreadLocal<Long> waitNanosLocal = new ThreadLocal<Long>();
private boolean logDifferentThread = true;
private volatile boolean keepAlive = false;
private SQLException initException = null;
private boolean asyncInit = false;
protected boolean killWhenSocketReadTimeout = false;
protected boolean checkExecuteTime = false;
@ -178,11 +180,11 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
protected static final AtomicLongFieldUpdater<DruidDataSource> createTaskIdSeedUpdater
= AtomicLongFieldUpdater.newUpdater(DruidDataSource.class, "createTaskIdSeed");
public DruidDataSource(){
public DruidDataSource() {
this(false);
}
public DruidDataSource(boolean fairLock){
public DruidDataSource(boolean fairLock) {
super(fairLock);
configFromPropety(System.getProperties());
@ -526,6 +528,45 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
}
}
private synchronized void doSomethingBeforeCreationThreadBreak() {
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();
if (destroyConnectionThread != null) {
if (!destroyConnectionThread.isInterrupted()) {
destroyConnectionThread.interrupt();
}
}
destroyConnectionThread = new DestroyConnectionThread(destroyName);
destroyConnectionThread.setStarted(false);
initedLatch = new CountDownLatch(2);
}
private void checkThread() throws SQLException {
if (createConnectionThread == null) {
throw new IllegalStateException("createConnectionThread not start!");
}
if (destroyConnectionThread == null) {
throw new IllegalStateException("destroyConnectionThread not start!");
}
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 boolean isKillWhenSocketReadTimeout() {
return killWhenSocketReadTimeout;
}
@ -798,6 +839,11 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
}
public void init() throws SQLException {
if (initException != null) {
LOG.error("{dataSource-" + this.getID() + "} init error", initException);
throw initException;
}
if (inited) {
return;
}
@ -944,6 +990,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
createAndLogThread();
createAndStartCreatorThread();
createAndStartDestroyThread();
createAndStartDetectThread();
initedLatch.await();
init = true;
@ -968,16 +1015,13 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
} catch (SQLException e) {
LOG.error("{dataSource-" + this.getID() + "} init error", e);
initException = e;
throw e;
} catch (InterruptedException e) {
throw new SQLException(e.getMessage(), e);
} catch (RuntimeException e){
LOG.error("{dataSource-" + this.getID() + "} init error", e);
throw e;
} catch (Error e){
LOG.error("{dataSource-" + this.getID() + "} init error", e);
} catch (Throwable e) {
initException = new SQLException(e.getMessage());
throw e;
} finally {
inited = true;
lock.unlock();
@ -1087,7 +1131,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;
@ -1096,6 +1140,15 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
initedLatch.countDown();
}
private void createAndStartDetectThread() {
if (createScheduler == null) {
String threadName = "Druid-ConnectionPool-Detection-" + System.identityHashCode(this) + this.getUrl();
periodDetectionThread = new PeriodDetectionThread(threadName);
periodDetectionThread.start();
}
}
/**
* load filters from SPI ServiceLoader
*
@ -1181,21 +1234,21 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
return;
}
String errorMessage = "";
String infoMessage = "";
if (testOnBorrow) {
errorMessage += "testOnBorrow is true, ";
if (isTestOnBorrow()) {
infoMessage += "testOnBorrow is true, ";
}
if (testOnReturn) {
errorMessage += "testOnReturn is true, ";
if (isTestOnReturn()) {
infoMessage += "testOnReturn is true, ";
}
if (testWhileIdle) {
errorMessage += "testWhileIdle is true, ";
if (isTestWhileIdle()) {
infoMessage += "testWhileIdle is true, ";
}
LOG.error(errorMessage + "validationQuery not set");
LOG.info(infoMessage + "validationQuery not set");
}
protected void resolveDriver() throws SQLException {
@ -1351,7 +1404,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
}
for (Class<?> driverClass = driver.getClass();;) {
for (Class<?> driverClass = driver.getClass(); ; ) {
String realDriverClassName = driverClass.getName();
if (realDriverClassName.equals(JdbcConstants.MYSQL_DRIVER) //
|| realDriverClassName.equals(JdbcConstants.MYSQL_DRIVER_6)) {
@ -1402,6 +1455,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);
@ -1422,7 +1476,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 {
@ -1591,7 +1645,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
DruidConnectionHolder holder;
for (boolean createDirect = false;;) {
for (boolean createDirect = false; ; ) {
if (createDirect) {
createStartNanosUpdater.set(this, System.nanoTime());
if (creatingCountUpdater.compareAndSet(this, 0, 1)) {
@ -1756,7 +1810,9 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
buf.append(", createErrorCount ").append(createErrorCount);
}
List<JdbcSqlStatValue> sqlList = this.getDataSourceStat().getRuningSqlList();
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');
@ -1764,10 +1820,12 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
buf.append(", ");
}
JdbcSqlStatValue sql = sqlList.get(i);
buf.append("runningSqlCount ").append(sql.getRunningCount());
buf.append("runningSqlCount ");
buf.append(sql.getRunningCount());
buf.append(" : ");
buf.append(sql.getSql());
}
}
String errorMessage = buf.toString();
@ -1858,13 +1916,12 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
lock.unlock();
}
if(onFatalError && holder != null && holder.getDataSource() != null) {
if (onFatalError && holder != null && holder.getDataSource() != null) {
ReentrantLock dataSourceLock = holder.getDataSource().lock;
dataSourceLock.lock();
try {
emptySignal();
}
finally {
} finally {
dataSourceLock.unlock();
}
}
@ -2097,6 +2154,10 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
destroyConnectionThread.interrupt();
}
if (periodDetectionThread != null) {
periodDetectionThread.interrupt();
}
if (createSchedulerFuture != null) {
createSchedulerFuture.cancel(true);
}
@ -2248,7 +2309,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
@ -2494,7 +2555,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
value.setCommitCount(commitCountUpdater.getAndSet(this, 0));
value.setRollbackCount(rollbackCountUpdater.getAndSet(this, 0));
value.setPstmtCacheHitCount(cachedPreparedStatementHitCountUpdater.getAndSet(this,0));
value.setPstmtCacheHitCount(cachedPreparedStatementHitCountUpdater.getAndSet(this, 0));
value.setPstmtCacheMissCount(cachedPreparedStatementMissCountUpdater.getAndSet(this, 0));
value.setStartTransactionCount(startTransactionCountUpdater.getAndSet(this, 0));
@ -2603,7 +2664,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
}
private void runInternal() {
for (;;) {
for (; ; ) {
// addLast
lock.lock();
@ -2769,8 +2830,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);
}
@ -2780,7 +2842,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
long lastDiscardCount = 0;
int errorCount = 0;
for (;;) {
for (; ; ) {
// addLast
try {
lock.lockInterruptibly();
@ -2829,6 +2891,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
if ((!closing) && (!closed)) {
LOG.error("create connection Thread Interrupted, url: " + jdbcUrl, e);
}
DruidDataSource.this.doSomethingBeforeCreationThreadBreak();
break;
} finally {
lock.unlock();
@ -2838,9 +2901,13 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
try {
connection = createPhysicalConnection();
} catch (SQLException e) {
LOG.error("create connection SQLException, url: " + jdbcUrl + ", errorCode " + e.getErrorCode()
+ ", state " + e.getSQLState(), e);
} catch (SQLException | RuntimeException e) {
if (e instanceof SQLException) {
LOG.error("create connection error, url: " + jdbcUrl + ", errorCode " + ((SQLException) e).getErrorCode()
+ ", state " + ((SQLException) e).getSQLState(), e);
} else {
LOG.error("create connection error", e);
}
errorCount++;
if (errorCount > connectionErrorRetryAttempts && timeBetweenConnectErrorMillis > 0) {
@ -2861,17 +2928,16 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
try {
Thread.sleep(timeBetweenConnectErrorMillis);
} catch (InterruptedException interruptEx) {
break;
} catch (InterruptedException ignore) {
}
DruidDataSource.this.doSomethingBeforeCreationThreadBreak();
break;
}
} catch (RuntimeException e) {
LOG.error("create connection RuntimeException", e);
setFailContinuous(true);
continue;
} catch (Error e) {
LOG.error("create connection Error", e);
setFailContinuous(true);
DruidDataSource.this.doSomethingBeforeCreationThreadBreak();
break;
}
@ -2892,11 +2958,49 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
}
}
}
public boolean isStarted() {
return started;
}
public void setStarted(boolean started) {
this.started = started;
}
}
//周期性检查生产线程状态,因为在终止生产线程的时候，为了不让生产线程疯狂重试数据库，只是生成了一个生产线程，但是并没有start,需要一个守护线程
//周期性检查线程状态，帮助其启动。
private class PeriodDetectionThread extends Thread {
public PeriodDetectionThread(String name) {
super(name);
this.setDaemon(true);
}
public void run() {
while (true) {
synchronized (DruidDataSource.this) {
//生产线程发生了切换，并且有线程在等待连接，需要主动唤醒生产线程，否则由getConnection方法来唤醒生产线程
if (!createConnectionThread.started && !destroyConnectionThread.started && notEmptyWaitThreadCount > 0) {
createConnectionThread.setStarted(true);
createConnectionThread.start();
destroyConnectionThread.setStarted(true);
destroyConnectionThread.start();
}
}
try {
Thread.sleep(30000);
} catch (InterruptedException ignore) {
break;
}
}
}
}
public class DestroyConnectionThread extends Thread {
public DestroyConnectionThread(String name){
private volatile boolean started = true;
public DestroyConnectionThread(String name) {
super(name);
this.setDaemon(true);
}
@ -2904,7 +3008,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
public void run() {
initedLatch.countDown();
for (;;) {
for (; ; ) {
// 从前面开始删除
try {
if (closed || closing) {
@ -2928,6 +3032,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 {
@ -2948,14 +3059,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) {
@ -2981,7 +3092,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()) {
@ -3049,7 +3160,9 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
return removeCount;
}
/** Instance key */
/**
* Instance key
*/
protected String instanceKey = null;
public Reference getReference() throws NamingException {
@ -3814,7 +3927,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
}
int fillCount = 0;
for (;;) {
for (; ; ) {
try {
lock.lockInterruptibly();
} catch (InterruptedException e) {

Write Preview
Loading…
Cancel
Save