摘要：基于擴(kuò)展實(shí)現(xiàn)真正的數(shù)據(jù)庫(kù)連接池這種方案中，項(xiàng)目占用的連接數(shù)僅僅為。一種是連接暫時(shí)不再使用，其占用狀態(tài)解除，可以從使用者手中交回到空閑隊(duì)列中這種我們稱為連接的歸隊(duì)。源碼剖析系列目錄

作者：bromine
鏈接：https://www.jianshu.com/p/1a7...
來(lái)源：簡(jiǎn)書(shū)
著作權(quán)歸作者所有，本文已獲得作者授權(quán)轉(zhuǎn)載，并對(duì)原文進(jìn)行了重新的排版。
Swoft Github: https://github.com/swoft-clou...

對(duì)于基于php-fpm的傳統(tǒng)php-web應(yīng)用，包括且不限于Mysql,Redis,RabbitMq，每次請(qǐng)求到來(lái)都需要為其新建一套獨(dú)享的的連接，這直接帶來(lái)了一些典型問(wèn)題：

連接開(kāi)銷:連接隨著http請(qǐng)求到來(lái)而新建,隨著請(qǐng)求返回而銷毀，大量連接新建銷毀是對(duì)系統(tǒng)資源的浪費(fèi)。

連接數(shù)量過(guò)高:每一個(gè)請(qǐng)求都需要一套自己的連接，系統(tǒng)連接數(shù)和并發(fā)數(shù)會(huì)成一個(gè)近線性的關(guān)系。如果系統(tǒng)并發(fā)量達(dá)到了1w,那么就需要建立1w個(gè)對(duì)應(yīng)的連接，這對(duì)于Mysql之類的后端服務(wù)而言，是一個(gè)大的負(fù)荷。

空閑連接:假設(shè)我們有一個(gè)接口使用了一個(gè)Mysql連接。該接口在一開(kāi)始進(jìn)行一次sql查詢后，后面的操作都是sql無(wú)關(guān)的，那么該請(qǐng)求占據(jù)的空閑連接完全就是一種資源的浪費(fèi)。

對(duì)于異步系統(tǒng)而言，這個(gè)問(wèn)題變得更加的嚴(yán)峻。一個(gè)請(qǐng)求處理進(jìn)程要對(duì)同一個(gè)服務(wù)進(jìn)行并發(fā)的操作，意味著這個(gè)請(qǐng)求要持有1個(gè)以上同類的連接，這對(duì)于系統(tǒng)壓力而言，無(wú)疑是雪上加霜了，所以連接池對(duì)于基于Swoole的Web框架而言已經(jīng)是一個(gè)必需實(shí)現(xiàn)的機(jī)制了。

連接池作為一個(gè)SCOPE為SINGLETON的典型Bean,
其實(shí)例最早會(huì)在SwoftBeanBeanFactory::reload()階段被初始化。

對(duì)于RPC或者HTTP請(qǐng)求而言，關(guān)系最密切的進(jìn)程肯定是Worker和Task進(jìn)程了。
對(duì)于這兩者而言 SwoftBeanBeanFactory::reload()會(huì)在swoole的onWorkerStart事件的回調(diào)階段階段被調(diào)用。

//SwoftBootstrapServerServerTrait(HttpServer和RpcServer都使用了該性狀)
/**
 * OnWorkerStart event callback
 *
 * @param Server $server server
 * @param int $workerId workerId
 * @throws InvalidArgumentException
 */
public function onWorkerStart(Server $server, int $workerId)
{
    // Init Worker and TaskWorker
    $setting = $server->setting;
    $isWorker = false;

    if ($workerId >= $setting["worker_num"]) {
        // TaskWorker
        ApplicationContext::setContext(ApplicationContext::TASK);
        ProcessHelper::setProcessTitle($this->serverSetting["pname"] . " task process");
    } else {
        // Worker
        $isWorker = true;
        ApplicationContext::setContext(ApplicationContext::WORKER);
        ProcessHelper::setProcessTitle($this->serverSetting["pname"] . " worker process");
    }

    $this->fireServerEvent(SwooleEvent::ON_WORKER_START, [$server, $workerId, $isWorker]);
    //beforeWorkerStart()內(nèi)部會(huì)調(diào)用BeanFactory::reload();
    $this->beforeWorkerStart($server, $workerId, $isWorker);
}

這意味著此時(shí)的連接池對(duì)象的生命周期是 進(jìn)程全局期而不是程序全局期
將進(jìn)程池設(shè)計(jì)為進(jìn)程全局期，而不是共享程度最高的程序全局期原因，個(gè)人認(rèn)為主要有3個(gè)

多個(gè)進(jìn)程同時(shí)對(duì)一個(gè)連接進(jìn)行讀寫會(huì)導(dǎo)致數(shù)據(jù)傳輸錯(cuò)亂，需要保證連接不會(huì)被同時(shí)訪問(wèn)。

Worker進(jìn)程對(duì)程序全局期的對(duì)象進(jìn)行寫操作時(shí)會(huì)導(dǎo)致寫時(shí)復(fù)制，產(chǎn)生一個(gè)進(jìn)程全局期的副本，程序全局期較難維持。

使用進(jìn)程全局期的話可以利用現(xiàn)有的Bean機(jī)制管理對(duì)象，減少的特殊編碼。

//SwoftProcessProcessBuilder.php
/**
     * After process
     *
     * @param string $processName
     * @param bool   $boot 該參數(shù)即Process 注解的boot屬性
     */
    private static function beforeProcess(string $processName, $boot)
    {
        if ($boot) {
            BeanFactory::reload();
            $initApplicationContext = new InitApplicationContext();
            $initApplicationContext->init();
        }

        App::trigger(ProcessEvent::BEFORE_PROCESS, null, $processName);
    }
}

Swoft中的Process有兩種：一種是定義Process 注解的boot屬性為true的 前置進(jìn)程，這種進(jìn)程隨系統(tǒng)啟動(dòng)而啟動(dòng)的 ；另一種是定義Process 注解的boot屬性為false的 用戶自定義進(jìn)程 ,該類進(jìn)程需要用戶在需要的時(shí)候手動(dòng)調(diào)用ProcessBuilder::create()啟動(dòng) 。

但是無(wú)論是何者，最終都會(huì)在Process中調(diào)用beforeProcess()進(jìn)行子進(jìn)程的初始化。對(duì)于 boot為true的 前置進(jìn)程 ，由于其啟動(dòng)時(shí)父進(jìn)程還未初始化bean容器，所以會(huì)多帶帶進(jìn)行bean容器初始化，而對(duì)于boot為false的其他 用戶自定義進(jìn)程,其會(huì)直接繼承父進(jìn)程的Ioc容器。

Swoft基本上遵守著一個(gè)進(jìn)程擁有一個(gè)多帶帶連接池的規(guī)則，這樣所有進(jìn)程中的連接都是獨(dú)立的，保證了連接
不會(huì)被同時(shí)讀寫。唯獨(dú)在Process中有一個(gè)特例。如果對(duì)先使用依賴連接池的服務(wù)，如對(duì)Mysql進(jìn)行CRUD，再調(diào)用ProcessBuilder::create()啟動(dòng) 用戶自定義進(jìn)程，由于用戶自定義進(jìn)程 會(huì)直接繼承父進(jìn)程的Bean容器而不重置，這時(shí)子進(jìn)程會(huì)獲得父進(jìn)程中的連接池和連接。

/**
 * The adapter of command
 * @Bean()
 */
class HandlerAdapter
{
    /**
     * before command
     *
     * @param string $class
     * @param string $command
     * @param bool   $server
     */
    private function beforeCommand(string $class, string $command, bool $server)
    {
        if ($server) {
            return;
        }
        $this->bootstrap();
        BeanFactory::reload();

        // 初始化
        $spanId = 0;
        $logId = uniqid();

        $uri = $class . "->" . $command;
        $contextData = [
            "logid"       => $logId,
            "spanid"      => $spanId,
            "uri"         => $uri,
            "requestTime" => microtime(true),
        ];

        RequestContext::setContextData($contextData);
    }
}

命令行腳本擁有自己多帶帶的Bean容器，其情況和Process相似且更簡(jiǎn)單，嚴(yán)格遵循一個(gè)進(jìn)程一個(gè)連接池，這里不再累述。

假設(shè)Worker數(shù)目為j,Task數(shù)目為k,Process數(shù)為l,Command數(shù)為m，每個(gè)進(jìn)程池內(nèi)配置最大連接數(shù)為n，部署機(jī)器數(shù)為x，不難看出每個(gè)swoft項(xiàng)目占用的連接數(shù)為(j+k+l+m)*n*x。

天峰本人曾經(jīng)提過(guò)另一種基于Swoole的連接池模型。
Rango-<基于swoole擴(kuò)展實(shí)現(xiàn)真正的PHP數(shù)據(jù)庫(kù)連接池>

這種方案中，項(xiàng)目占用的連接數(shù)僅僅為k*x。
除了Task進(jìn)程各個(gè)進(jìn)程并不直接持有連接池，而是通過(guò)向Task進(jìn)程提交指令(task(),sendMessage())讓其代為進(jìn)行連接池相關(guān)服務(wù)的操作，至少需要額外的一次進(jìn)程間通信（默認(rèn)為Unix Socket）。
該方案雖然能夠更好的復(fù)用連接和節(jié)省連接數(shù)，但機(jī)制實(shí)現(xiàn)并不方便。從另一個(gè)角度去看，Swoft的連接池方案是為了解決使用Swoole時(shí)，單進(jìn)程并發(fā)執(zhí)行的連接數(shù)要求問(wèn)題；Range提出的連接池方案是為了解決超大流量系統(tǒng)下對(duì)Mysql等服務(wù)的壓力控制問(wèn)題。兩者適合不同的場(chǎng)景，其目的和意義在一定程度下是重合的，但并不是完全一樣的。

連接池根據(jù)當(dāng)前是否協(xié)程環(huán)境選擇一種合適的隊(duì)列結(jié)構(gòu)作為連接的容器。

SplQueue:SplQueue是PHP標(biāo)準(zhǔn)庫(kù)的數(shù)據(jù)結(jié)構(gòu)，底層是一個(gè)雙向鏈表,在隊(duì)列操作這種特化場(chǎng)景下，性能遠(yuǎn)高于底層使用鏈表+哈希表實(shí)現(xiàn)的array()數(shù)據(jù)結(jié)構(gòu)。

SwooleCoroutineChannel是Swoole提供的協(xié)程相關(guān)的數(shù)據(jù)結(jié)構(gòu)，不僅提供了常規(guī)的隊(duì)列操作。在協(xié)程環(huán)境下，當(dāng)其隊(duì)列長(zhǎng)度從0至1之間切換時(shí)，會(huì)自動(dòng)讓出協(xié)程控制權(quán)并喚醒對(duì)應(yīng)的生產(chǎn)者或消費(fèi)者。

SwoftPoolConnectionPool.php
abstract class ConnectionPool implements PoolInterface {
    /**
     * Get connection
     *
     * @throws ConnectionException;
     * @return ConnectionInterface
     */
    public function getConnection():ConnectionInterface
    {
        //根據(jù)執(zhí)行環(huán)境選擇容器
        if (App::isCoContext()) {
            $connection = $this->getConnectionByChannel();
        } else {
            $connection = $this->getConnectionByQueue();
        }

        //連接使用前的檢查和重新連接
        if ($connection->check() == false) {
            $connection->reconnect();
        }
        //加入到全局上下文中，事務(wù)處理和資源相關(guān)的監(jiān)聽(tīng)事件會(huì)用到
        $this->addContextConnection($connection);
        return $connection;
    }
}

SwoftPoolConnectionPool.php
/**
 * Get connection by queue
 *
 * @return ConnectionInterface
 * @throws ConnectionException
 */
private function getConnectionByQueue(): ConnectionInterface
{
    if($this->queue == null){
        $this->queue = new SplQueue();
    }
    
    if (!$this->queue->isEmpty()) {
        //隊(duì)列存在可用連接直接獲取
        return $this->getEffectiveConnection($this->queue->count(), false);
    }
    //超出隊(duì)列最大長(zhǎng)度
    if ($this->currentCount >= $this->poolConfig->getMaxActive()) {
        throw new ConnectionException("Connection pool queue is full");
    }
    //向隊(duì)列補(bǔ)充連接
    $connect = $this->createConnection();
    $this->currentCount++;

    return $connect;
}

SwoftPoolConnectionPool.php
/**
 * Get effective connection
 *
 * @param int  $queueNum
 * @param bool $isChannel
 *
 * @return ConnectionInterface
 */
private function getEffectiveConnection(int $queueNum, bool $isChannel = true): ConnectionInterface
{
    $minActive = $this->poolConfig->getMinActive();
    //連接池中連接少于數(shù)量下限時(shí)直接獲取
    if ($queueNum <= $minActive) {
        return $this->getOriginalConnection($isChannel);
    }

    $time        = time();
    $moreActive  = $queueNum - $minActive;
    $maxWaitTime = $this->poolConfig->getMaxWaitTime();
    //檢查多余的連接，如等待時(shí)間過(guò)長(zhǎng)，表示當(dāng)前所持連接數(shù)暫時(shí)大于需求值，且易失效，直接釋放
    for ($i = 0; $i < $moreActive; $i++) {
        /* @var ConnectionInterface $connection */
        $connection = $this->getOriginalConnection($isChannel);;
        $lastTime = $connection->getLastTime();
        if ($time - $lastTime < $maxWaitTime) {
            return $connection;
        }
        $this->currentCount--;
    }

    return $this->getOriginalConnection($isChannel);
}

加點(diǎn)注釋就非常清晰了，此處不再贅述。

連接的釋放有兩種不同的容易引起歧義的用法，為此我們做以下定義：
一種是連接已經(jīng)不再使用了，可以關(guān)閉了，這種我們稱為 連接的銷毀。
一種是連接暫時(shí)不再使用，其占用狀態(tài)解除，可以從使用者手中交回到空閑隊(duì)列中,這種我們稱為 連接的歸隊(duì)。

一般通過(guò)unset變量，或者通過(guò)其他手段清除連接變量的所有引用，等待Zend引擎實(shí)現(xiàn)鏈接資源清理。
這一點(diǎn)在上文的getEffectiveConnection()中出現(xiàn)過(guò)。執(zhí)行到$this->currentCount--;的時(shí)候 ，連接已經(jīng)出隊(duì)了，而$connection變量會(huì)在下個(gè)循環(huán)時(shí)作為循環(huán)變量被替換或者方法返回時(shí)作為局部變量被清除，連接資源的引用清0.引用降到0的資源會(huì)在下次gc執(zhí)行時(shí)被回收，所以你沒(méi)看到主動(dòng)的連接釋放代碼也很正常。
如果你的代碼在其他地方引用了這連接而沒(méi)管理好，可能會(huì)導(dǎo)致資源泄露。

/**
 * Class AbstractConnect
 */
abstract class AbstractConnection implements ConnectionInterface
{
    //SwoftPoolAbstractConnection.php
    /**
     * @param bool $release
     */
    public function release($release = false)
    {
        if ($this->isAutoRelease() || $release) {
            $this->pool->release($this);
        }
    }
}

//SwoftPoolConnectionPool.php
/**
 * Class ConnectPool
 */
abstract class ConnectionPool implements PoolInterface
{
    /**
     * Release connection
     *
     * @param ConnectionInterface $connection
     */
    public function release(ConnectionInterface $connection)
    {
        $connectionId = $connection->getConnectionId();
        $connection->updateLastTime();
        $connection->setRecv(true);
        $connection->setAutoRelease(true);

        if (App::isCoContext()) {
            $this->releaseToChannel($connection);
        } else {
            $this->releaseToQueue($connection);
        }

        $this->removeContextConnection($connectionId);
    }
}

當(dāng)用戶使用完某個(gè)連接后，比如執(zhí)行了完了一條sql后，應(yīng)當(dāng)調(diào)用連接的release()方法。
連接本身是持有連接池的反向連接，在用戶調(diào)用ConnectionInterface->release()方法時(shí)，并不會(huì)馬上銷毀自身，而是清理自身的標(biāo)記，調(diào)用PoolInterface->release()重新加入到連接池中。

//SwoftEventListenersResourceReleaseListener.php
/**
 * Resource release listener
 *
 * @Listener(AppEvent::RESOURCE_RELEASE)
 */
class ResourceReleaseListener implements EventHandlerInterface
{
    /**
     * @param SwoftEventEventInterface $event
     * @throws InvalidArgumentException
     */
    public function handle(EventInterface $event)
    {
        // Release system resources
        App::trigger(AppEvent::RESOURCE_RELEASE_BEFORE);

        $connectionKey = PoolHelper::getContextCntKey();
        $connections   = RequestContext::getContextDataByKey($connectionKey, []);
        if (empty($connections)) {
            return;
        }

        /* @var SwoftPoolConnectionInterface $connection */
        foreach ($connections as $connectionId => $connection) {
            if (!$connection->isRecv()) {
                Log::error(sprintf("%s connection is not received ，forget to getResult()", get_class($connection)));
                $connection->receive();
            }

            Log::error(sprintf("%s connection is not released ，forget to getResult()", get_class($connection)));
            $connection->release(true);
        }
    }
}

考慮到用戶可能會(huì)在使用完后沒(méi)有釋放連接造成連接泄露，Swoft會(huì)在Rpc/Http請(qǐng)求或者Task結(jié)束后觸發(fā)一個(gè)Swoft.resourceRelease事件（注:Swoft是筆者添加的前綴，方便讀者區(qū)分Swoole相關(guān)事件和Swoft相關(guān)事件)，將連接強(qiáng)制收包并歸隊(duì)。

Swoft源碼剖析系列目錄：https://segmentfault.com/a/11...