天天看點

搞懂Druid之連接配接建立和銷毀

作者:Java架構學習指南

前言

Druid是阿裡開源的資料庫連接配接池,是阿裡監控系統Dragoon的副産品,提供了強大的可監控性和基于Filter-Chain的可擴充性。

本篇文章将對Druid資料庫連接配接池的連接配接建立和銷毀進行分析。分析Druid資料庫連接配接池的源碼前,需要明确幾個概念。

  1. Druid資料庫連接配接池中可用的連接配接存放在一個數組connections中;
  2. Druid資料庫連接配接池做并發控制,主要靠一把可重入鎖以及和這把鎖關聯的兩個Condition對象;
public DruidAbstractDataSource(boolean lockFair) {
   lock = new ReentrantLock(lockFair);

   notEmpty = lock.newCondition();
   empty = lock.newCondition();
}           
  1. 連接配接池沒有可用連接配接時,應用線程會在notEmpty上等待,連接配接池已滿時,生産連接配接的線程會在empty上等待;
  2. 對連接配接保活,就是每間隔一定時間,對達到了保活間隔周期的連接配接進行有效性校驗,可以将無效連接配接銷毀,也可以防止連接配接長時間不與資料庫服務端通信。

Druid版本:1.2.11

正文

一. DruidDataSource連接配接建立

DruidDataSource連接配接的建立由CreateConnectionThread線程完成,其run() 方法如下所示。

public void run() {
    initedLatch.countDown();

    long lastDiscardCount = 0;
    int errorCount = 0;
    for (; ; ) {
        try {
            lock.lockInterruptibly();
        } catch (InterruptedException e2) {
            break;
        }

        long discardCount = DruidDataSource.this.discardCount;
        boolean discardChanged = discardCount - lastDiscardCount > 0;
        lastDiscardCount = discardCount;

        try {
            // emptyWait為true表示生産連接配接線程需要等待,無需生産連接配接
            boolean emptyWait = true;

            // 發生了建立錯誤,且池中已無連接配接,且丢棄連接配接的統計沒有改變
            // 此時生産連接配接線程需要生産連接配接
            if (createError != null
                    && poolingCount == 0
                    && !discardChanged) {
                emptyWait = false;
            }

            if (emptyWait
                    && asyncInit && createCount < initialSize) {
                emptyWait = false;
            }

            if (emptyWait) {
                // 池中已有連接配接數大于等于正在等待連接配接的應用線程數
                // 且目前是非keepAlive場景
                // 且目前是非連續失敗
                // 此時生産連接配接的線程在empty上等待
                // keepAlive && activeCount + poolingCount < minIdle時會在shrink()方法中觸發emptySingal()來添加連接配接
                // isFailContinuous()傳回true表示連續失敗,即多次(預設2次)建立實體連接配接失敗
                if (poolingCount >= notEmptyWaitThreadCount
                        && (!(keepAlive && activeCount + poolingCount < minIdle))
                        && !isFailContinuous()
                ) {
                    empty.await();
                }

                // 防止建立超過maxActive數量的連接配接
                if (activeCount + poolingCount >= maxActive) {
                    empty.await();
                    continue;
                }
            }

        } catch (InterruptedException e) {
            // 省略
        } finally {
            lock.unlock();
        }

        PhysicalConnectionInfo connection = null;

        try {
            connection = createPhysicalConnection();
        } catch (SQLException e) {
            LOG.error("create connection SQLException, url: " + jdbcUrl
                    + ", errorCode " + e.getErrorCode()
                    + ", state " + e.getSQLState(), e);

            errorCount++;
            if (errorCount > connectionErrorRetryAttempts
                    && timeBetweenConnectErrorMillis > 0) {
                // 多次建立失敗
                setFailContinuous(true);
                // 如果配置了快速失敗,就喚醒所有在notEmpty上等待的應用線程
                if (failFast) {
                    lock.lock();
                    try {
                        notEmpty.signalAll();
                    } finally {
                        lock.unlock();
                    }
                }

                if (breakAfterAcquireFailure) {
                    break;
                }

                try {
                    Thread.sleep(timeBetweenConnectErrorMillis);
                } catch (InterruptedException interruptEx) {
                    break;
                }
            }
        } catch (RuntimeException e) {
            LOG.error("create connection RuntimeException", e);
            setFailContinuous(true);
            continue;
        } catch (Error e) {
            LOG.error("create connection Error", e);
            setFailContinuous(true);
            break;
        }

        if (connection == null) {
            continue;
        }

        // 把連接配接添加到連接配接池
        boolean result = put(connection);
        if (!result) {
            JdbcUtils.close(connection.getPhysicalConnection());
            LOG.info("put physical connection to pool failed.");
        }

        errorCount = 0;

        if (closing || closed) {
            break;
        }
    }
}           

CreateConnectionThread的run() 方法整體就是在一個死循環中不斷的等待,被喚醒,然後建立線程。當一個實體連接配接被建立出來後,會調用DruidDataSource#put方法将其放到連接配接池connections中,put() 方法源碼如下所示。

protected boolean put(PhysicalConnectionInfo physicalConnectionInfo) {
    DruidConnectionHolder holder = null;
    try {
        holder = new DruidConnectionHolder(DruidDataSource.this, physicalConnectionInfo);
    } catch (SQLException ex) {
        // 省略
        return false;
    }

    return put(holder, physicalConnectionInfo.createTaskId, false);
}

private boolean put(DruidConnectionHolder holder,
                    long createTaskId, boolean checkExists) {
    // 涉及到連接配接池中連接配接數量改變的操作,都需要加鎖
    lock.lock();
    try {
        if (this.closing || this.closed) {
            return false;
        }

        // 池中已有連接配接數已經大于等于最大連接配接數,則不再把連接配接加到連接配接池并直接傳回false
        if (poolingCount >= maxActive) {
            if (createScheduler != null) {
                clearCreateTask(createTaskId);
            }
            return false;
        }

        // 檢查重複添加
        if (checkExists) {
            for (int i = 0; i < poolingCount; i++) {
                if (connections[i] == holder) {
                    return false;
                }
            }
        }

        // 連接配接放入連接配接池
        connections[poolingCount] = holder;
        // poolingCount++
        incrementPoolingCount();

        if (poolingCount > poolingPeak) {
            poolingPeak = poolingCount;
            poolingPeakTime = System.currentTimeMillis();
        }

        // 喚醒在notEmpty上等待連接配接的應用線程
        notEmpty.signal();
        notEmptySignalCount++;

        if (createScheduler != null) {
            clearCreateTask(createTaskId);

            if (poolingCount + createTaskCount < notEmptyWaitThreadCount
                    && activeCount + poolingCount + createTaskCount < maxActive) {
                emptySignal();
            }
        }
    } finally {
        lock.unlock();
    }
    return true;
}           

put() 方法會先将實體連接配接從PhysicalConnectionInfo中擷取出來并封裝成一個DruidConnectionHolder,DruidConnectionHolder就是Druid連接配接池中的連接配接。新添加的連接配接會存放在連接配接池數組connections的poolingCount位置,然後poolingCount會加1,也就是poolingCount代表着連接配接池中可以擷取的連接配接的數量。

二. DruidDataSource連接配接銷毀

DruidDataSource連接配接的銷毀由DestroyConnectionThread線程完成,其run() 方法如下所示。

public void run() {
    // run()方法隻要執行了,就調用initedLatch#countDown
    initedLatch.countDown();

    for (; ; ) {
        // 每間隔timeBetweenEvictionRunsMillis執行一次DestroyTask的run()方法
        try {
            if (closed || closing) {
                break;
            }

            if (timeBetweenEvictionRunsMillis > 0) {
                Thread.sleep(timeBetweenEvictionRunsMillis);
            } else {
                Thread.sleep(1000);
            }

            if (Thread.interrupted()) {
                break;
            }

            // 執行DestroyTask的run()方法來銷毀需要銷毀的連接配接
            destroyTask.run();
        } catch (InterruptedException e) {
            break;
        }
    }
}           

DestroyConnectionThread的run() 方法就是在一個死循環中每間隔timeBetweenEvictionRunsMillis的時間就執行一次DestroyTask的run() 方法。DestroyTask#run方法實作如下所示。

public void run() {
    // 根據一系列條件判斷并銷毀連接配接
    shrink(true, keepAlive);

    // RemoveAbandoned機制
    if (isRemoveAbandoned()) {
        removeAbandoned();
    }
}           

在DestroyTask#run方法中會調用DruidDataSource#shrink方法來根據設定的條件來判斷出需要銷毀和保活的連接配接。DruidDataSource#shrink方法如下所示。

// checkTime參數表示在将一個連接配接進行銷毀前,是否需要判斷一下空閑時間
public void shrink(boolean checkTime, boolean keepAlive) {
    // 加鎖
    try {
        lock.lockInterruptibly();
    } catch (InterruptedException e) {
        return;
    }

    // needFill = keepAlive && poolingCount + activeCount < minIdle
    // needFill為true時,會調用empty.signal()喚醒生産連接配接的線程來生産連接配接
    boolean needFill = false;
    // evictCount記錄需要銷毀的連接配接數
    // keepAliveCount記錄需要保活的連接配接數
    int evictCount = 0;
    int keepAliveCount = 0;
    int fatalErrorIncrement = fatalErrorCount - fatalErrorCountLastShrink;
    fatalErrorCountLastShrink = fatalErrorCount;

    try {
        if (!inited) {
            return;
        }

        // checkCount = 池中已有連接配接數 - 最小空閑連接配接數
        // 正常情況下,最多能夠将前checkCount個連接配接進行銷毀
        final int checkCount = poolingCount - minIdle;
        final long currentTimeMillis = System.currentTimeMillis();
        // 正常情況下,需要周遊池中所有連接配接
        // 從前往後周遊,i為數組索引
        for (int i = 0; i < poolingCount; ++i) {
            DruidConnectionHolder connection = connections[i];

            // 如果發生了緻命錯誤(onFatalError == true)且緻命錯誤發生時間(lastFatalErrorTimeMillis)在連接配接建立時間之後
            // 把連接配接加入到保活連接配接數組中
            if ((onFatalError || fatalErrorIncrement > 0)
                    && (lastFatalErrorTimeMillis > connection.connectTimeMillis)) {
                keepAliveConnections[keepAliveCount++] = connection;
                continue;
            }

            if (checkTime) {
                // phyTimeoutMillis表示連接配接的實體存活逾時時間,預設值是-1
                if (phyTimeoutMillis > 0) {
                    // phyConnectTimeMillis表示連接配接的實體存活時間
                    long phyConnectTimeMillis = currentTimeMillis
                            - connection.connectTimeMillis;
                    // 連接配接的實體存活時間大于phyTimeoutMillis,則将這個連接配接放入evictConnections數組
                    if (phyConnectTimeMillis > phyTimeoutMillis) {
                        evictConnections[evictCount++] = connection;
                        continue;
                    }
                }

                // idleMillis表示連接配接的空閑時間
                long idleMillis = currentTimeMillis - connection.lastActiveTimeMillis;

                // minEvictableIdleTimeMillis表示連接配接允許的最小空閑時間,預設是30分鐘
                // keepAliveBetweenTimeMillis表示保活間隔時間,預設是2分鐘
                // 如果連接配接的空閑時間小于minEvictableIdleTimeMillis且還小于keepAliveBetweenTimeMillis
                // 則connections數組中目前連接配接之後的連接配接都會滿足空閑時間小于minEvictableIdleTimeMillis且還小于keepAliveBetweenTimeMillis
                // 此時跳出周遊,不再檢查其餘的連接配接
                if (idleMillis < minEvictableIdleTimeMillis
                        && idleMillis < keepAliveBetweenTimeMillis
                ) {
                    break;
                }

                // 連接配接的空閑時間大于等于允許的最小空閑時間
                if (idleMillis >= minEvictableIdleTimeMillis) {
                    if (checkTime && i < checkCount) {
                        // i < checkCount這個條件的了解如下:
                        // 每次shrink()方法執行時,connections數組中隻有索引0到checkCount-1的連接配接才允許被銷毀
                        // 這樣才能保證銷毀完連接配接後,connections數組中至少還有minIdle個連接配接
                        evictConnections[evictCount++] = connection;
                        continue;
                    } else if (idleMillis > maxEvictableIdleTimeMillis) {
                        // 如果空閑時間過久,已經大于了允許的最大空閑時間(預設7小時)
                        // 那麼無論如何都要銷毀這個連接配接
                        evictConnections[evictCount++] = connection;
                        continue;
                    }
                }

                // 如果開啟了保活機制,且連接配接空閑時間大于等于了保活間隔時間
                // 此時将連接配接加入到保活連接配接數組中
                if (keepAlive && idleMillis >= keepAliveBetweenTimeMillis) {
                    keepAliveConnections[keepAliveCount++] = connection;
                }
            } else {
                // checkTime為false,那麼前checkCount個連接配接直接進行銷毀,不再判斷這些連接配接的空閑時間是否超過門檻值
                if (i < checkCount) {
                    evictConnections[evictCount++] = connection;
                } else {
                    break;
                }
            }
        }

        // removeCount = 銷毀連接配接數 + 保活連接配接數
        // removeCount表示本次從connections數組中拿掉的連接配接數
        // 注:一定是從前往後拿,正常情況下最後minIdle個連接配接是安全的
        int removeCount = evictCount + keepAliveCount;
        if (removeCount > 0) {
            // [0, 1, 2, 3, 4, null, null, null] -> [3, 4, 2, 3, 4, null, null, null]
            System.arraycopy(connections, removeCount, connections, 0, poolingCount - removeCount);
            // [3, 4, 2, 3, 4, null, null, null] -> [3, 4, null, null, null, null, null, null, null]
            Arrays.fill(connections, poolingCount - removeCount, poolingCount, null);
            // 更新池中連接配接數
            poolingCount -= removeCount;
        }
        keepAliveCheckCount += keepAliveCount;

        // 如果池中連接配接數加上活躍連接配接數(借出去的連接配接)小于最小空閑連接配接數
        // 則将needFill設為true,後續需要喚醒生産連接配接的線程來生産連接配接
        if (keepAlive && poolingCount + activeCount < minIdle) {
            needFill = true;
        }
    } finally {
        lock.unlock();
    }

    if (evictCount > 0) {
        // 周遊evictConnections數組,銷毀其中的連接配接
        for (int i = 0; i < evictCount; ++i) {
            DruidConnectionHolder item = evictConnections[i];
            Connection connection = item.getConnection();
            JdbcUtils.close(connection);
            destroyCountUpdater.incrementAndGet(this);
        }
        Arrays.fill(evictConnections, null);
    }

    if (keepAliveCount > 0) {
        // 周遊keepAliveConnections數組,對其中的連接配接做可用性校驗
        // 校驗通過連接配接就放入connections數組,沒通過連接配接就銷毀
        for (int i = keepAliveCount - 1; i >= 0; --i) {
            DruidConnectionHolder holer = keepAliveConnections[i];
            Connection connection = holer.getConnection();
            holer.incrementKeepAliveCheckCount();

            boolean validate = false;
            try {
                this.validateConnection(connection);
                validate = true;
            } catch (Throwable error) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug("keepAliveErr", error);
                }
            }

            boolean discard = !validate;
            if (validate) {
                holer.lastKeepTimeMillis = System.currentTimeMillis();
                boolean putOk = put(holer, 0L, true);
                if (!putOk) {
                    discard = true;
                }
            }

            if (discard) {
                try {
                    connection.close();
                } catch (Exception e) {

                }

                lock.lock();
                try {
                    discardCount++;

                    if (activeCount + poolingCount <= minIdle) {
                        emptySignal();
                    }
                } finally {
                    lock.unlock();
                }
            }
        }
        this.getDataSourceStat().addKeepAliveCheckCount(keepAliveCount);
        Arrays.fill(keepAliveConnections, null);
    }

    // 如果needFill為true則喚醒生産連接配接的線程來生産連接配接
    if (needFill) {
        lock.lock();
        try {
            // 計算需要生産連接配接的個數
            int fillCount = minIdle - (activeCount + poolingCount + createTaskCount);
            for (int i = 0; i < fillCount; ++i) {
                emptySignal();
            }
        } finally {
            lock.unlock();
        }
    } else if (onFatalError || fatalErrorIncrement > 0) {
        lock.lock();
        try {
            emptySignal();
        } finally {
            lock.unlock();
        }
    }
}           

在DruidDataSource#shrink方法中,核心邏輯是周遊connections數組中的連接配接,并判斷這些連接配接是需要銷毀還是需要保活。通常情況下,connections數組中的前checkCount(checkCount = poolingCount - minIdle) 個連接配接是危險的,因為這些連接配接隻要滿足了:空閑時間 >= minEvictableIdleTimeMillis(允許的最小空閑時間),那麼就需要被銷毀,而connections數組中的最後minIdle個連接配接是相對安全的,因為這些連接配接隻有在滿足:空閑時間 > maxEvictableIdleTimeMillis(允許的最大空閑時間) 時,才會被銷毀。這麼判斷的原因,主要就是需要讓連接配接池裡能夠保證至少有minIdle個空閑連接配接可以讓應用線程擷取。

當确定好了需要銷毀和需要保活的連接配接後,此時會先将connections數組清理,隻保留安全的連接配接,這個過程示意圖如下。

搞懂Druid之連接配接建立和銷毀

最後,會周遊evictConnections數組,銷毀數組中的連接配接,周遊keepAliveConnections數組,對其中的每個連接配接做可用性校驗,如果校驗可用,那麼就重新放回connections數組,否則銷毀。

總結

連接配接的建立由一個叫做CreateConnectionThread的線程完成,整體流程就是在一個死循環中不斷的等待,被喚醒,然後建立連接配接。每一個被建立出來的實體連接配接java.sql.Connection會被封裝為一個DruidConnectionHolder,然後存放到connections數組中。

連接配接的銷毀由一個叫做DestroyConnectionThread的線程完成,核心邏輯是周期性的周遊connections數組中的連接配接,并判斷這些連接配接是需要銷毀還是需要保活,需要銷毀的連接配接最後會被實體銷毀,需要保活的連接配接最後會進行一次可用性校驗,如果校驗不通過,則進行實體銷毀。