support create direct.

2023-03-04 19:17:16 +00:00 · 2017-11-20 22:55:23 +08:00
parent f6863f013b
commit 3fb484aa88
3 changed files with 236 additions and 65 deletions
@@ -37,9 +37,7 @@ import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.*;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.logging.Logger;
@@ -235,9 +233,10 @@ public abstract class DruidAbstractDataSource extends WrapperAdapter implements

    protected ReentrantLock                            activeConnectionLock                      = new ReentrantLock();

-    protected AtomicInteger                            creatingCount                             = new AtomicInteger();
+    protected volatile int                             creatingCount                             = 0;
    protected AtomicLong                               createCount                               = new AtomicLong();
    protected AtomicLong                               destroyCount                              = new AtomicLong();
+    protected volatile int                             directCreateCount                         = 0;

    private Boolean                                    useUnfairLock                             = null;

@@ -258,6 +257,9 @@ public abstract class DruidAbstractDataSource extends WrapperAdapter implements
    protected volatile boolean                         onFatalError                              = false;
    protected volatile int                             onFatalErrorMaxActive                     = 0;

+    protected static AtomicIntegerFieldUpdater<DruidAbstractDataSource> directCreateCountUpdater    = AtomicIntegerFieldUpdater.newUpdater(DruidAbstractDataSource.class, "directCreateCount");
+    protected static AtomicIntegerFieldUpdater<DruidAbstractDataSource> createCountUpdater          = AtomicIntegerFieldUpdater.newUpdater(DruidAbstractDataSource.class, "creatingCount");
+
    public DruidAbstractDataSource(boolean lockFair){
        lock = new ReentrantLock(lockFair);

@@ -1562,7 +1564,7 @@ public abstract class DruidAbstractDataSource extends WrapperAdapter implements
                ? new HashMap<String, Object>()
                : null;

-        creatingCount.incrementAndGet();
+        createCountUpdater.incrementAndGet(this);
        try {
            conn = createPhysicalConnection(url, physicalConnectProperties);
            connectedNanos = System.nanoTime();
@@ -1594,7 +1596,7 @@ public abstract class DruidAbstractDataSource extends WrapperAdapter implements
        } finally {
            long nano = System.nanoTime() - connectStartNanos;
            createTimespan += nano;
-            creatingCount.decrementAndGet();
+            createCountUpdater.decrementAndGet(this);
        }

        return new PhysicalConnectionInfo(conn, connectStartNanos, connectedNanos, initedNanos, validatedNanos, variables, globalVariables);
@@ -24,11 +24,10 @@ import java.sql.Connection;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.*;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Future;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;

@@ -93,10 +92,11 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
    private final static Log                 LOG                       = LogFactory.getLog(DruidDataSource.class);
    private static final long                serialVersionUID          = 1L;
    // stats
-    private final AtomicLong                 recycleErrorCount         = new AtomicLong();
+    private volatile long                    recycleErrorCount         = 0L;
    private long                             connectCount              = 0L;
    private long                             closeCount                = 0L;
-    private final AtomicLong                 connectErrorCount         = new AtomicLong();
+    private volatile long                    connectErrorCount         = 0L;
+//    private final AtomicLong                 connectErrorCount         = new AtomicLong();
    private long                             recycleCount              = 0L;
    private long                             removeAbandonedCount      = 0L;
    private long                             notEmptyWaitCount         = 0L;
@@ -152,6 +152,11 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
    private boolean                          asyncInit                 = false;
    protected boolean                        killWhenSocketReadTimeout = false;

+    protected static final AtomicLongFieldUpdater<DruidDataSource> recycleErrorCountUpdater
+            = AtomicLongFieldUpdater.newUpdater(DruidDataSource.class, "recycleErrorCount");
+    protected static final AtomicLongFieldUpdater<DruidDataSource> connectErrorCountUpdater
+            = AtomicLongFieldUpdater.newUpdater(DruidDataSource.class, "connectErrorCount");
+
    public DruidDataSource(){
        this(false);
    }
@@ -548,7 +553,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
            lock.unlock();
        }

-        connectErrorCount.set(0);
+        connectErrorCountUpdater.set(this, 0);
        errorCount.set(0);
        commitCount.set(0);
        rollbackCount.set(0);
@@ -558,7 +563,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
        preparedStatementCount.set(0);
        transactionHistogram.reset();
        cachedPreparedStatementDeleteCount.set(0);
-        recycleErrorCount.set(0);
+        recycleErrorCountUpdater.set(this, 0);

        resetCount.incrementAndGet();
    }
@@ -1341,12 +1346,12 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat

    private DruidPooledConnection getConnectionInternal(long maxWait) throws SQLException {
        if (closed) {
-            connectErrorCount.incrementAndGet();
+            connectErrorCountUpdater.incrementAndGet(this);
            throw new DataSourceClosedException("dataSource already closed at " + new Date(closeTimeMillis));
        }

        if (!enable) {
-            connectErrorCount.incrementAndGet();
+            connectErrorCountUpdater.incrementAndGet(this);
            throw new DataSourceDisableException();
        }

@@ -1354,52 +1359,104 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
        final int maxWaitThreadCount = this.maxWaitThreadCount;

        DruidConnectionHolder holder;
-        try {
-            lock.lockInterruptibly();
-        } catch (InterruptedException e) {
-            connectErrorCount.incrementAndGet();
-            throw new SQLException("interrupt", e);
-        }

-        try {
-            if (maxWaitThreadCount > 0
-                    && notEmptyWaitThreadCount >= maxWaitThreadCount)
-            {
-                    connectErrorCount.incrementAndGet();
-                    throw new SQLException("maxWaitThreadCount " + maxWaitThreadCount + ", current wait Thread count "
-                                           + lock.getQueueLength());
-            }
+        for (boolean createDirect = false;;) {
+            if (createDirect) {
+                if (createCountUpdater.compareAndSet(this, 0, 1)) {
+                    PhysicalConnectionInfo pyConnInfo = DruidDataSource.this.createPhysicalConnection();
+                    holder = new DruidConnectionHolder(this, pyConnInfo);
+                    holder.lastActiveTimeMillis = System.currentTimeMillis();

-            if (onFatalError
-                    && onFatalErrorMaxActive > 0
-                    && activeCount >= onFatalErrorMaxActive) {
-                connectErrorCount.incrementAndGet();
-                throw new SQLException("onFatalError, activeCount " + activeCount + ", onFatalErrorMaxActive " + onFatalErrorMaxActive);
-            }
+                    createCountUpdater.decrementAndGet(this);
+                    directCreateCountUpdater.incrementAndGet(this);

-            connectCount++;
+                    if (LOG.isDebugEnabled()) {
+                        LOG.debug("conn-direct_create ");
+                    }

-            if (maxWait > 0) {
-                holder = pollLast(nanos);
-            } else {
-                holder = takeLast();
-            }
+                    boolean discard = false;
+                    lock.lock();
+                    try {
+                        if (activeCount < maxActive) {
+                            activeCount++;
+                            if (activeCount > activePeak) {
+                                activePeak = activeCount;
+                                activePeakTime = System.currentTimeMillis();
+                            }
+                            break;
+                        } else {
+                            discard = true;
+                        }
+                    } finally {
+                        lock.unlock();
+                    }

-            if (holder != null) {
-                activeCount++;
-                if (activeCount > activePeak) {
-                    activePeak = activeCount;
-                    activePeakTime = System.currentTimeMillis();
+                    if (discard) {
+                        JdbcUtils.close(pyConnInfo.getPhysicalConnection());
+                    }
                }
            }
-        } catch (InterruptedException e) {
-            connectErrorCount.incrementAndGet();
-            throw new SQLException(e.getMessage(), e);
-        } catch (SQLException e) {
-            connectErrorCount.incrementAndGet();
-            throw e;
-        } finally {
-            lock.unlock();
+
+            try {
+                lock.lockInterruptibly();
+            } catch (InterruptedException e) {
+                connectErrorCountUpdater.incrementAndGet(this);
+                throw new SQLException("interrupt", e);
+            }
+
+            try {
+                if (maxWaitThreadCount > 0
+                        && notEmptyWaitThreadCount >= maxWaitThreadCount) {
+                    connectErrorCountUpdater.incrementAndGet(this);
+                    throw new SQLException("maxWaitThreadCount " + maxWaitThreadCount + ", current wait Thread count "
+                            + lock.getQueueLength());
+                }
+
+                if (onFatalError
+                        && onFatalErrorMaxActive > 0
+                        && activeCount >= onFatalErrorMaxActive) {
+                    connectErrorCountUpdater.incrementAndGet(this);
+                    throw new SQLException("onFatalError, activeCount " + activeCount + ", onFatalErrorMaxActive " + onFatalErrorMaxActive);
+                }
+
+                connectCount++;
+
+                if (createScheduler != null
+                        && poolingCount == 0
+                        && activeCount < maxActive
+                        && createCountUpdater.get(this) == 0
+                        && createScheduler instanceof ScheduledThreadPoolExecutor) {
+                    ScheduledThreadPoolExecutor executor = (ScheduledThreadPoolExecutor) createScheduler;
+                    if (executor.getQueue().size() > 0) {
+                        createDirect = true;
+                        continue;
+                    }
+                }
+
+                if (maxWait > 0) {
+                    holder = pollLast(nanos);
+                } else {
+                    holder = takeLast();
+                }
+
+                if (holder != null) {
+                    activeCount++;
+                    if (activeCount > activePeak) {
+                        activePeak = activeCount;
+                        activePeakTime = System.currentTimeMillis();
+                    }
+                }
+            } catch (InterruptedException e) {
+                connectErrorCountUpdater.incrementAndGet(this);
+                throw new SQLException(e.getMessage(), e);
+            } catch (SQLException e) {
+                connectErrorCountUpdater.incrementAndGet(this);
+                throw e;
+            } finally {
+                lock.unlock();
+            }
+
+            break;
        }

        if (holder == null) {
@@ -1410,7 +1467,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
               .append(waitNanos / (1000 * 1000))//
               .append(", active ").append(activeCount)//
               .append(", maxActive ").append(maxActive)//
-               .append(", creating ").append(creatingCount.get())//
+               .append(", creating ").append(creatingCount)//
            ;

            List<JdbcSqlStatValue> sqlList = this.getDataSourceStat().getRuningSqlList();
@@ -1629,12 +1686,12 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
            }

            LOG.error("recyle error", e);
-            recycleErrorCount.incrementAndGet();
+            recycleErrorCountUpdater.incrementAndGet(this);
        }
    }

    public long getRecycleErrorCount() {
-        return recycleErrorCount.get();
+        return recycleErrorCountUpdater.get(this);
    }

    public void clearStatementCache() throws SQLException {
@@ -1804,7 +1861,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
                notEmptyWaitCount++;

                if (!enable) {
-                    connectErrorCount.incrementAndGet();
+                    connectErrorCountUpdater.incrementAndGet(this);
                    throw new DataSourceDisableException();
                }
            }
@@ -1851,7 +1908,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
                    notEmptyWaitNanos += (startEstimate - estimate);

                    if (!enable) {
-                        connectErrorCount.incrementAndGet();
+                        connectErrorCountUpdater.incrementAndGet(this);
                        throw new DataSourceDisableException();
                    }
                } catch (InterruptedException ie) {
@@ -1936,7 +1993,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
    }

    public long getConnectErrorCount() {
-        return connectErrorCount.get();
+        return connectErrorCountUpdater.get(this);
    }

    @Override
@@ -2054,7 +2111,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
        }
        value.setDefaultTransactionIsolation(this.getDefaultTransactionIsolation());

-        value.setLogicConnectErrorCount(connectErrorCount.getAndSet(0));
+        value.setLogicConnectErrorCount(connectErrorCountUpdater.getAndSet(this, 0));

        value.setPhysicalConnectCount(createCount.getAndSet(0));
        value.setPhysicalCloseCount(destroyCount.getAndSet(0));
@@ -3260,7 +3317,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
            try {
                lock.lockInterruptibly();
            } catch (InterruptedException e) {
-                connectErrorCount.incrementAndGet();
+                connectErrorCountUpdater.incrementAndGet(this);
                throw new SQLException("interrupt", e);
            }

@@ -3278,14 +3335,14 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
                holder = new DruidConnectionHolder(this, pyConnInfo);
            } catch (SQLException e) {
                LOG.error("fill connection error, url: " + this.jdbcUrl, e);
-                connectErrorCount.incrementAndGet();
+                connectErrorCountUpdater.incrementAndGet(this);
                throw e;
            }

            try {
                lock.lockInterruptibly();
            } catch (InterruptedException e) {
-                connectErrorCount.incrementAndGet();
+                connectErrorCountUpdater.incrementAndGet(this);
                throw new SQLException("interrupt", e);
            }

@@ -0,0 +1,112 @@
+package com.alibaba.druid.bvt.pool;
+
+import com.alibaba.druid.PoolTestCase;
+import com.alibaba.druid.mock.MockConnection;
+import com.alibaba.druid.mock.MockDriver;
+import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.druid.util.JdbcUtils;
+import org.junit.Assert;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
+public class CreateSchedulerTest_directCreate extends PoolTestCase {
+
+    private DruidDataSource[]        dataSources;
+    private ScheduledExecutorService createScheduler;
+    private ScheduledExecutorService destroyScheduler;
+
+    protected void setUp() throws Exception {
+        super.setUp();
+
+        createScheduler = Executors.newScheduledThreadPool(1);
+        destroyScheduler = Executors.newScheduledThreadPool(1);
+
+        dataSources = new DruidDataSource[8];
+        for (int i = 0; i < dataSources.length; ++i) {
+            DruidDataSource dataSource = new DruidDataSource();
+            dataSource.setUrl("jdbc:mock:xxx");
+            dataSource.setDriver(new SlowDriver());
+            dataSource.setCreateScheduler(createScheduler);
+            dataSource.setDestroyScheduler(destroyScheduler);
+            dataSource.setMinEvictableIdleTimeMillis(10);
+            dataSource.setTimeBetweenEvictionRunsMillis(10);
+            dataSource.setFilters("log4j");
+            dataSource.setName("ds_" + i);
+
+            dataSources[i] = dataSource;
+        }
+    }
+
+    protected void tearDown() throws Exception {
+        for (int i = 0; i < dataSources.length; ++i) {
+            JdbcUtils.close(dataSources[i]);
+        }
+
+        createScheduler.shutdown();
+
+        super.tearDown();
+    }
+
+    public void test_connectAndClose() throws Exception {
+        Thread[] threads = new Thread[32];
+        final CountDownLatch latch = new CountDownLatch(threads.length);
+        for (int i = 0; i < threads.length; ++i) {
+            final int threadId = i;
+            int dataSourceIndex = i % dataSources.length;
+            final DruidDataSource dataSource = dataSources[dataSourceIndex];
+            Thread thread = new Thread(new Task(dataSource, latch), "thread-" + threadId);
+            threads[i] = thread;
+        }
+
+        for (int i = 0; i < threads.length; ++i) {
+            threads[i].start();
+            Thread.sleep(10);
+        }
+
+        latch.await();
+    }
+
+    public static class SlowDriver extends MockDriver {
+        public MockConnection createMockConnection(MockDriver driver, String url, Properties connectProperties) {
+            try {
+                Thread.sleep(1000 * 1);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+            return super.createMockConnection(driver, url, connectProperties);
+        }
+    }
+
+    public static class Task implements Runnable {
+        private final DruidDataSource dataSource;
+        private final CountDownLatch latch;
+
+        public Task(DruidDataSource dataSource, CountDownLatch latch) {
+            this.dataSource = dataSource;
+            this.latch = latch;
+        }
+
+        @Override
+        public void run() {
+            try {
+                for (int i = 0; i < 4; ++i) {
+                    Connection conn = null;
+                    try {
+                        conn = dataSource.getConnection();
+                    } catch (SQLException e) {
+                        e.printStackTrace();
+                    } finally {
+                        JdbcUtils.close(conn);
+                    }
+                }
+            } finally {
+                latch.countDown();
+            }
+        }
+    }
+}