druid断线重连机制修复

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

@@ -35,6 +35,7 @@ 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;
 
@@ -143,7 +144,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
     private LogStatsThread logStatsThread;
     private int createTaskCount;
 
-    private final CountDownLatch             initedLatch             = new CountDownLatch(2);
+    private CountDownLatch initedLatch = new CountDownLatch(2);
 
     private volatile boolean enable = true;
 
@@ -167,11 +168,11 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
 
     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 +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;
@@ -817,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;
@@ -1055,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);
@@ -1076,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 {
@@ -1235,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);
@@ -1249,7 +1274,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) {
@@ -1633,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
 
@@ -1946,7 +1971,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
         }
 
         private void runInternal() {
-            for (;;) {
+            for (; ; ) {
 
                 // addLast
                 lock.lock();
@@ -2043,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);
         }
@@ -2054,7 +2080,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
 
             long lastDiscardCount = 0;
             int errorCount = 0;
-            for (;;) {
+            for (; ; ) {
                 // addLast
                 try {
                     lock.lockInterruptibly();
@@ -2126,6 +2152,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 +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);
         }
@@ -2162,7 +2199,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
         public void run() {
             initedLatch.countDown();
 
-            for (;;) {
+            for (; ; ) {
                 // 从前面开始删除
                 try {
                     if (closed) {
@@ -2186,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 {
@@ -2203,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) {
@@ -2236,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()) {
@@ -2304,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 {
@@ -2998,7 +3044,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
         }
 
         int fillCount = 0;
-        for (;;) {
+        for (; ; ) {
             try {
                 lock.lockInterruptibly();
             } catch (InterruptedException e) {