java連接池Druid獲取連接getConnection示例詳解

更新時(shí)間：2023年09月15日 11:51:53 作者：福

這篇文章主要為大家介紹了java連接池Druid獲取連接getConnection示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪

Druid連接池

Druid連接池只存儲在connections數(shù)組中，所以獲取連接的邏輯應(yīng)該比HikariPool簡單一些：直接從connectoins獲取即可。

DruidDataSource.getConnection

直接上代碼：

@Override
    public DruidPooledConnection getConnection() throws SQLException {
        return getConnection(maxWait);
    }

調(diào)用了getConnection(maxWait)，maxWait是參數(shù)設(shè)定的獲取連接的最長等待時(shí)間，超過該時(shí)長還沒有獲取到連接的話，拋異常。

看getConnection(maxWait)代碼：

public DruidPooledConnection getConnection(long maxWaitMillis) throws SQLException {
        init();
        if (filters.size() > 0) {
            FilterChainImpl filterChain = new FilterChainImpl(this);
            return filterChain.dataSource_connect(this, maxWaitMillis);
        } else {
            return getConnectionDirect(maxWaitMillis);
        }
    }

先調(diào)用init，init方法會判斷連接池是否已經(jīng)完成了初始化，如果沒有完成初始化則首先進(jìn)行初始化，初始化的代碼我們上一篇文章已經(jīng)分析過了。

之后判斷是否有filters，filters的內(nèi)容我們先放放，暫時(shí)不管，直接看沒有filters的情況下，調(diào)用getConnectionDirect方法。

getConnectionDirect

方法比較長，我們還是老辦法，分段分析：

public DruidPooledConnection getConnectionDirect(long maxWaitMillis) throws SQLException {
        int notFullTimeoutRetryCnt = 0;
        for (;;) {
            // handle notFullTimeoutRetry
            DruidPooledConnection poolableConnection;
            try {
                poolableConnection = getConnectionInternal(maxWaitMillis);
            } catch (GetConnectionTimeoutException ex) {
                if (notFullTimeoutRetryCnt <= this.notFullTimeoutRetryCount && !isFull()) {
                    notFullTimeoutRetryCnt++;
                    if (LOG.isWarnEnabled()) {
                        LOG.warn("get connection timeout retry : " + notFullTimeoutRetryCnt);
                    }
                    continue;
                }
                throw ex;
            }

上來之后首先無限for循環(huán)，目的是從連接池獲取到連接之后，根據(jù)參數(shù)設(shè)定可能會做必要的檢查，如果檢查不通過（比如連接不可用、連接已關(guān)閉等等）的話循環(huán)重新獲取。

然后調(diào)用getConnectionInternal獲取連接，getConnectionInternal方法應(yīng)該是我們今天文章的主角，我們稍微放一放，為了文章的可讀性，先分析完getConnectionDirect方法。

我們假設(shè)通過調(diào)用getConnectionInternal方法獲取到一個(gè)連接（注意獲取到的連接對象是DruidPooledConnection，不是Connection對象，這個(gè)也不難想象，連接池獲取到的連接一定是數(shù)據(jù)庫物理連接的代理對象（或者叫封裝對象，封裝了數(shù)據(jù)庫物理連接Connection對象的對象，這個(gè)原理我們在分析HikariPool的時(shí)候已經(jīng)說過了。這個(gè)DruidPooledConnection對象我們也暫時(shí)放一放，后面分析）。

調(diào)用getConnectionInternal方法如果返回超時(shí)異常，判斷：如果當(dāng)前連接池沒滿，而且獲取連接超時(shí)重試次數(shù)小于參數(shù)notFullTimeoutRetryCount設(shè)定的次數(shù)的話，則continue，重新獲取連接。否則，拋出超時(shí)異常。

接下來：

if (testOnBorrow) {
                boolean validate = testConnectionInternal(poolableConnection.holder, poolableConnection.conn);
                if (!validate) {
                    if (LOG.isDebugEnabled()) {
                        LOG.debug("skip not validate connection.");
                    }
                    discardConnection(poolableConnection.holder);
                    continue;
                }
            } else {
                if (poolableConnection.conn.isClosed()) {
                    discardConnection(poolableConnection.holder); // 傳入null，避免重復(fù)關(guān)閉
                    continue;
                }

testOnBorrow參數(shù)的目的是：獲取連接后是否要做連接可用性測試，如果設(shè)定為true的話，調(diào)用testConnectionInternal測試連接的可用性，testConnectionInternal方法上一篇文章分析連接回收的時(shí)候、處理keepAlive的過程中就碰到過，就是執(zhí)行配置好的sql語句測試連接可用性，如果測試不通過的話則調(diào)用discardConnection關(guān)閉連接，continue重新獲取連接。

否則，如果testOnBorrow參數(shù)沒有打開的話，檢查當(dāng)前連接如果已經(jīng)關(guān)閉，則調(diào)用discardConnection關(guān)閉連接（沒太明白連接既然已經(jīng)是關(guān)閉狀態(tài)，為啥還需要調(diào)用？），continue重新獲取連接。

不建議打開testOnBorrow參數(shù)，因?yàn)檫B接池都會有連接回收機(jī)制，比如上一篇文章講過的Druid的DestroyConnectionThread & DestroyTask，回收參數(shù)配置正常的話，回收機(jī)制基本可以確保連接的可用性。打開testOnBorrow參數(shù)會導(dǎo)致每次獲取連接之后都測試連接的可用性，嚴(yán)重影響系統(tǒng)性能。

接下來：

if (testWhileIdle) {
                    final DruidConnectionHolder holder = poolableConnection.holder;
                    long currentTimeMillis             = System.currentTimeMillis();
                    long lastActiveTimeMillis          = holder.lastActiveTimeMillis;
                    long lastExecTimeMillis            = holder.lastExecTimeMillis;
                    long lastKeepTimeMillis            = holder.lastKeepTimeMillis;
                    if (checkExecuteTime
                            && lastExecTimeMillis != lastActiveTimeMillis) {
                        lastActiveTimeMillis = lastExecTimeMillis;
                    }
                    if (lastKeepTimeMillis > lastActiveTimeMillis) {
                        lastActiveTimeMillis = lastKeepTimeMillis;
                    }
                    long idleMillis                    = currentTimeMillis - lastActiveTimeMillis;
                    long timeBetweenEvictionRunsMillis = this.timeBetweenEvictionRunsMillis;
                    if (timeBetweenEvictionRunsMillis <= 0) {
                        timeBetweenEvictionRunsMillis = DEFAULT_TIME_BETWEEN_EVICTION_RUNS_MILLIS;
                    }
                    if (idleMillis >= timeBetweenEvictionRunsMillis
                            || idleMillis < 0 // unexcepted branch
                            ) {
                        boolean validate = testConnectionInternal(poolableConnection.holder, poolableConnection.conn);
                        if (!validate) {
                            if (LOG.isDebugEnabled()) {
                                LOG.debug("skip not validate connection.");
                            }
                            discardConnection(poolableConnection.holder);
                             continue;
                        }
                    }
                }
            }

這段代碼的邏輯是：參數(shù)testWhileIdle設(shè)置為true的話，檢查當(dāng)前鏈接的空閑時(shí)長如果大于timeBetweenEvictionRunsMillis（默認(rèn)60秒）的話，則調(diào)用testConnectionInternal測試連接可用性，連接不可用則關(guān)閉連接，continue重新獲取連接。

然后：

if (removeAbandoned) {
                StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace();
                poolableConnection.connectStackTrace = stackTrace;
                poolableConnection.setConnectedTimeNano();
                poolableConnection.traceEnable = true;
                activeConnectionLock.lock();
                try {
                    activeConnections.put(poolableConnection, PRESENT);
                } finally {
                    activeConnectionLock.unlock();
                }
            }

出現(xiàn)了一個(gè)removeAbandoned參數(shù)，這個(gè)參數(shù)的意思是移除被遺棄的連接對象，如果打開的話就把當(dāng)前連接放到activeConnections中，篇幅有限，這部分內(nèi)容就不展開了，后面我們會專門寫一篇文章介紹removeAbandoned參數(shù)。

剩下的一小部分代碼，很簡單，根據(jù)參數(shù)設(shè)置連接的autoCommit，之后返回連接poolableConnection。

if (!this.defaultAutoCommit) {
                poolableConnection.setAutoCommit(false);
            }
            return poolableConnection;
        }
    }

getConnectionDirect方法源碼分析完成了，下面我們要看一下getConnectionInternal方法，這是真正從連接池中獲取連接的方法。

getConnectionInternal

直接看代碼：

private DruidPooledConnection getConnectionInternal(long maxWait) throws SQLException {
        if (closed) {
            connectErrorCountUpdater.incrementAndGet(this);
            throw new DataSourceClosedException("dataSource already closed at " + new Date(closeTimeMillis));
        }
        if (!enable) {
            connectErrorCountUpdater.incrementAndGet(this);
            if (disableException != null) {
                throw disableException;
            }
            throw new DataSourceDisableException();
        }
        final long nanos = TimeUnit.MILLISECONDS.toNanos(maxWait);
        final int maxWaitThreadCount = this.maxWaitThreadCount;
        DruidConnectionHolder holder;

檢查連接池狀態(tài)如果已經(jīng)disable或cloesed的話，拋異常。

接下來：

for (boolean createDirect = false;;) {
            if (createDirect) {
                createStartNanosUpdater.set(this, System.nanoTime());
                if (creatingCountUpdater.compareAndSet(this, 0, 1)) {
                    PhysicalConnectionInfo pyConnInfo = DruidDataSource.this.createPhysicalConnection();
                    holder = new DruidConnectionHolder(this, pyConnInfo);
                    holder.lastActiveTimeMillis = System.currentTimeMillis();
                    creatingCountUpdater.decrementAndGet(this);
                    directCreateCountUpdater.incrementAndGet(this);
                    if (LOG.isDebugEnabled()) {
                        LOG.debug("conn-direct_create ");
                    }
                    boolean discard = false;
                    lock.lock();
                    try {
                        if (activeCount < maxActive) {
                            activeCount++;
                            holder.active = true;
                            if (activeCount > activePeak) {
                                activePeak = activeCount;
                                activePeakTime = System.currentTimeMillis();
                            }
                            break;
                        } else {
                            discard = true;
                        }
                    } finally {
                        lock.unlock();
                    }
                    if (discard) {
                        JdbcUtils.close(pyConnInfo.getPhysicalConnection());
                    }
                }
            }

初始化createDirect變量為false之后啟動無限循環(huán)，意思是不斷循環(huán)直到獲取到連接、或超時(shí)等其他異常情況發(fā)生。

緊接著的這段代碼是createDirect=true的情況下執(zhí)行的，createDirect是在下面循環(huán)體中檢查如果：createScheduler不為空、連接池空、活動連接數(shù)小于設(shè)定的最大活動連接數(shù)maxActive、并且createScheduler的隊(duì)列中排隊(duì)等待創(chuàng)建連接的線程大于0的情況下，設(shè)置createDirect為true的，以上這些條件如果成立的話，大概率表明createScheduler中的創(chuàng)建線程出問題了、所以createScheduler大概率指望不上了，所以要直接創(chuàng)建連接了。

直接創(chuàng)建的代碼也很容易理解，調(diào)用createPhysicalConnection創(chuàng)建物理連接，創(chuàng)建DruidConnectionHolder封裝該物理連接，創(chuàng)建之后獲取鎖資源，檢查activeCount < maxActive則表明創(chuàng)建連接成功、結(jié)束for循環(huán)，否則，activeCount >= maxActive則說明違反了原則（直接創(chuàng)建連接的過程中createScheduler可能復(fù)活了、又創(chuàng)建出來連接放入連接池中了），所以，關(guān)閉鎖資源之后，將剛創(chuàng)建出來的連接關(guān)閉。

然后：

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);
                    StringBuilder errorMsg = new StringBuilder();
                    errorMsg.append("onFatalError, activeCount ")
                            .append(activeCount)
                            .append(", onFatalErrorMaxActive ")
                            .append(onFatalErrorMaxActive);
                    if (lastFatalErrorTimeMillis > 0) {
                        errorMsg.append(", time '")
                                .append(StringUtils.formatDateTime19(
                                        lastFatalErrorTimeMillis, TimeZone.getDefault()))
                                .append("'");
                    }
                    if (lastFatalErrorSql != null) {
                        errorMsg.append(", sql \n")
                                .append(lastFatalErrorSql);
                    }
                    throw new SQLException(
                            errorMsg.toString(), lastFatalError);
                }
                connectCount++;
                if (createScheduler != null
                        && poolingCount == 0
                        && activeCount < maxActive
                        && creatingCountUpdater.get(this) == 0
                        && createScheduler instanceof ScheduledThreadPoolExecutor) {
                    ScheduledThreadPoolExecutor executor = (ScheduledThreadPoolExecutor) createScheduler;
                    if (executor.getQueue().size() > 0) {
                        createDirect = true;
                        continue;
                    }
                }

獲取鎖資源，檢查等待獲取連接的線程數(shù)如果大于參數(shù)設(shè)置的最大等待線程數(shù)，拋異常。

檢查并處理異常。

累加connectCount。

之后是上面提到過的對createDirect的處理。

接下來到了最為關(guān)鍵的部分，一般情況下createDirect為false，不會直接創(chuàng)建連接，邏輯會走到下面這部分代碼中，從連接池中獲取連接：

if (maxWait > 0) {
                    holder = pollLast(nanos);
                } else {
                    holder = takeLast();
                }
                if (holder != null) {
                    if (holder.discard) {
                        continue;
                    }
                    activeCount++;
                    holder.active = true;
                    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();
            }

如果參數(shù)設(shè)置了maxWait，則調(diào)用pollLast限時(shí)獲取，否則調(diào)用takeLast獲取連接，這兩個(gè)方法稍后分析。

之后檢查獲取到的連接已經(jīng)被discard的話，continue重新獲取連接。

釋放鎖資源。

從連接池中獲取到了連接，結(jié)束for循環(huán)。

如果takeLast或poolLast返回的DruidConnectionHolder為null的話（調(diào)用poolLast超時(shí)），處理錯(cuò)誤信息，拋GetConnectionTimeoutException超時(shí)異常（這部分代碼沒有貼出，省略了......感興趣的童鞋自己打開源碼看一下）。

否則，用DruidConnectionHolder封裝創(chuàng)建DruidPooledConnection后返回。

takeLast & pollLast(nanos)

這兩個(gè)方法的邏輯其實(shí)差不多，主要區(qū)別一個(gè)是限時(shí)，一個(gè)不限時(shí)，兩個(gè)方法都是在鎖狀態(tài)下執(zhí)行。

具體調(diào)用哪一個(gè)方法取決于參數(shù)maxWait，默認(rèn)值為-1，默認(rèn)情況下會調(diào)用takeLast，獲取連接的時(shí)候不限時(shí)。

建議設(shè)置maxWait，否則在特殊情況下如果創(chuàng)建連接失敗、會導(dǎo)致應(yīng)用層線程掛起，獲取不到任何返回的情況出現(xiàn)。如果設(shè)置了maxWait，getConnection方法會調(diào)用pollLast(nanos)，獲取不到連接后，應(yīng)用層會得到連接超時(shí)的反饋。

先看takeLast方法：

takeLast() throws InterruptedException, SQLException {
        try {
            while (poolingCount == 0) {
                emptySignal(); // send signal to CreateThread create connection
                if (failFast && isFailContinuous()) {
                    throw new DataSourceNotAvailableException(createError);
                }
                notEmptyWaitThreadCount++;
                if (notEmptyWaitThreadCount > notEmptyWaitThreadPeak) {
                    notEmptyWaitThreadPeak = notEmptyWaitThreadCount;
                }
                try {
                    notEmpty.await(); // signal by recycle or creator
                } finally {
                    notEmptyWaitThreadCount--;
                }
                notEmptyWaitCount++;
                if (!enable) {
                    connectErrorCountUpdater.incrementAndGet(this);
                    if (disableException != null) {
                        throw disableException;
                    }
                    throw new DataSourceDisableException();
                }
            }
        } catch (InterruptedException ie) {
            notEmpty.signal(); // propagate to non-interrupted thread
            notEmptySignalCount++;
            throw ie;
        }
        decrementPoolingCount();
        DruidConnectionHolder last = connections[poolingCount];
        connections[poolingCount] = null;
        return last;
    }

如果連接池為空(poolingCount == 0)的話，無限循環(huán)。

調(diào)用emptySignal()，通知創(chuàng)建連接線程，有人在等待獲取連接，抓緊時(shí)間創(chuàng)建連接。

然后調(diào)用notEmpty.await()，等待創(chuàng)建連接線程在完成創(chuàng)建、或者有連接歸還到連接池中后喚醒通知。

如果發(fā)生異常，調(diào)用一下notEmpty.signal()通知其他獲取連接的線程，沒準(zhǔn)自己沒能獲取成功、其他線程能獲取成功。

下面的代碼，線程池一定不空了。

線程池的線程數(shù)量減1（decrementPoolingCount），然后獲取connections的最后一個(gè)元素返回。

pollLast方法的代碼邏輯和takeLast的類似，只不過線程池空的話，當(dāng)前線程會限時(shí)掛起等待，超時(shí)仍然不能獲取到連接的話，直接返回null。

Druid連接池獲取連接代碼分析完畢！