Swoft 源碼剖析 - 鏈接池

時間 2019-11-05

標籤 swoft 源碼剖析鏈接简体版

原文原文鏈接

做者：bromine
連接：https://www.jianshu.com/p/1a7...
來源：簡書
著做權歸做者全部，本文已得到做者受權轉載，並對原文進行了從新的排版。
Swoft Github: https://github.com/swoft-clou...php

爲何須要引入鏈接池?

對於基於php-fpm的傳統php-web應用，包括且不限於Mysql,Redis,RabbitMq，每次請求到來都須要爲其新建一套獨享的的鏈接，這直接帶來了一些典型問題：git

鏈接開銷:鏈接隨着http請求到來而新建,隨着請求返回而銷燬，大量鏈接新建銷燬是對系統資源的浪費。
鏈接數量太高:每個請求都須要一套本身的鏈接，系統鏈接數和併發數會成一個近線性的關係。若是系統併發量達到了1w,那麼就須要創建1w個對應的鏈接，這對於Mysql之類的後端服務而言，是一個大的負荷。
空閒鏈接:假設咱們有一個接口使用了一個Mysql鏈接。該接口在一開始進行一次sql查詢後，後面的操做都是sql無關的，那麼該請求佔據的空閒鏈接徹底就是一種資源的浪費。

對於異步系統而言，這個問題變得更加的嚴峻。一個請求處理進程要對同一個服務進行併發的操做，意味着這個請求要持有1個以上同類的鏈接，這對於系統壓力而言，無疑是雪上加霜了，因此鏈接池對於基於Swoole的Web框架而言已是一個必需實現的機制了。github

Swoft鏈接池的生命週期與進程模型

鏈接池做爲一個SCOPE爲SINGLETON的典型Bean,
其實例最先會在Swoft\Bean\BeanFactory::reload()階段被初始化。web

Worker/Task進程

對於RPC或者HTTP請求而言，關係最密切的進程確定是Worker和Task進程了。
對於這二者而言 SwoftBeanBeanFactory::reload()會在swoole的onWorkerStart事件的回調階段階段被調用。sql

//Swoft\Bootstrap\Server\ServerTrait(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()內部會調用BeanFactory::reload();
    $this->beforeWorkerStart($server, $workerId, $isWorker);
}

這意味着此時的鏈接池對象的生命週期是 進程全局期而不是程序全局期
將進程池設計爲進程全局期，而不是共享程度最高的程序全局期緣由，我的認爲主要有3個數據庫

多個進程同時對一個鏈接進行讀寫會致使數據傳輸錯亂，須要保證鏈接不會被同時訪問。
Worker進程對程序全局期的對象進行寫操做時會致使寫時複製，產生一個進程全局期的副本，程序全局期較難維持。
使用進程全局期的話能夠利用現有的Bean機制管理對象，減小的特殊編碼。

Process中的鏈接池

//Swoft\Process\ProcessBuilder.php
/**
     * After process
     *
     * @param string $processName
     * @param bool   $boot 該參數即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的 前置進程，這種進程隨系統啓動而啓動的；另外一種是定義Process 註解的boot屬性爲false的 用戶自定義進程 ,該類進程須要用戶在須要的時候手動調用ProcessBuilder::create()啓動。bootstrap

可是不管是何者，最終都會在Process中調用beforeProcess()進行子進程的初始化。對於 boot爲true的 前置進程 ，因爲其啓動時父進程還未初始化bean容器，因此會單獨進行bean容器初始化，而對於boot爲false的其餘 用戶自定義進程,其會直接繼承父進程的Ioc容器。segmentfault

Swoft基本上遵照着一個進程擁有一個單獨鏈接池的規則，這樣全部進程中的鏈接都是獨立的，保證了鏈接
不會被同時讀寫。惟獨在Process中有一個特例。若是對先使用依賴鏈接池的服務，如對Mysql進行CRUD，再調用ProcessBuilder::create()啓動 用戶自定義進程，因爲用戶自定義進程 會直接繼承父進程的Bean容器而不重置，這時子進程會得到父進程中的鏈接池和鏈接。後端

Command

/**
 * 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類似且更簡單，嚴格遵循一個進程一個鏈接池，這裏再也不累述。swoole

假設Worker數目爲j,Task數目爲k,Process數爲l,Command數爲m，每一個進程池內配置最大鏈接數爲n，部署機器數爲x，不難看出每一個swoft項目佔用的鏈接數爲(j+k+l+m)*n*x。

天峯本人曾經提過另外一種基於Swoole的鏈接池模型。
Rango-<基於swoole擴展實現真正的PHP數據庫鏈接池>

這種方案中，項目佔用的鏈接數僅僅爲k*x。
除了Task進程各個進程並不直接持有鏈接池，而是經過向Task進程提交指令(task(),sendMessage())讓其代爲進行鏈接池相關服務的操做，至少須要額外的一次進程間通訊（默認爲Unix Socket）。
該方案雖然可以更好的複用鏈接和節省鏈接數，但機制實現並不方便。從另外一個角度去看，Swoft的鏈接池方案是爲了解決使用Swoole時，單進程併發執行的鏈接數要求問題；Range提出的鏈接池方案是爲了解決超大流量系統下對Mysql等服務的壓力控制問題。二者適合不一樣的場景，其目的和意義在必定程度下是重合的，但並非徹底同樣的。

Swoft鏈接池的實現

池的容器

鏈接池根據當前是否協程環境選擇一種合適的隊列結構做爲鏈接的容器。

\SplQueue:SplQueue是PHP標準庫的數據結構，底層是一個雙向鏈表,在隊列操做這種特化場景下，性能遠高於底層使用鏈表+哈希表實現的array()數據結構。
\Swoole\Coroutine\Channel是Swoole提供的協程相關的數據結構，不只提供了常規的隊列操做。在協程環境下，當其隊列長度從0至1之間切換時，會自動讓出協程控制權並喚醒對應的生產者或消費者。

鏈接的獲取

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

        //鏈接使用前的檢查和從新鏈接
        if ($connection->check() == false) {
            $connection->reconnect();
        }
        //加入到全局上下文中，事務處理和資源相關的監聽事件會用到
        $this->addContextConnection($connection);
        return $connection;
    }
}

\\Swoft\Pool\ConnectionPool.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()) {
        //隊列存在可用鏈接直接獲取
        return $this->getEffectiveConnection($this->queue->count(), false);
    }
    //超出隊列最大長度
    if ($this->currentCount >= $this->poolConfig->getMaxActive()) {
        throw new ConnectionException('Connection pool queue is full');
    }
    //向隊列補充鏈接
    $connect = $this->createConnection();
    $this->currentCount++;

    return $connect;
}

\\Swoft\Pool\ConnectionPool.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();
    //鏈接池中鏈接少於數量下限時直接獲取
    if ($queueNum <= $minActive) {
        return $this->getOriginalConnection($isChannel);
    }

    $time        = time();
    $moreActive  = $queueNum - $minActive;
    $maxWaitTime = $this->poolConfig->getMaxWaitTime();
    //檢查多餘的鏈接，如等待時間過長，表示當前所持鏈接數暫時大於需求值，且易失效，直接釋放
    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);
}

加點註釋就很是清晰了，此處再也不贅述。

鏈接的釋放

鏈接的釋放有兩種不一樣的容易引發歧義的用法，爲此咱們作如下定義：
一種是鏈接已經再也不使用了，能夠關閉了，這種咱們稱爲 鏈接的銷燬。
一種是鏈接暫時再也不使用，其佔用狀態解除，能夠從使用者手中交回到空閒隊列中,這種咱們稱爲 鏈接的歸隊。

連接的銷燬

通常經過unset變量，或者經過其餘手段清除鏈接變量的全部引用，等待Zend引擎實現連接資源清理。
這一點在上文的getEffectiveConnection()中出現過。執行到$this->currentCount--;的時候，鏈接已經出隊了，而$connection變量會在下個循環時做爲循環變量被替換或者方法返回時做爲局部變量被清除，鏈接資源的引用清0.引用降到0的資源會在下次gc執行時被回收，因此你沒看到主動的鏈接釋放代碼也很正常。
若是你的代碼在其餘地方引用了這鏈接而沒管理好，可能會致使資源泄露。

連接的歸隊

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

//Swoft\Pool\ConnectionPool.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);
    }
}

當用戶使用完某個鏈接後，好比執行了完了一條sql後，應當調用鏈接的release()方法。
鏈接自己是持有鏈接池的反向鏈接，在用戶調用ConnectionInterface->release()方法時，並不會立刻銷燬自身，而是清理自身的標記，調用PoolInterface->release()從新加入到鏈接池中。

//Swoft\Event\Listeners\ResourceReleaseListener.php
/**
 * Resource release listener
 *
 * @Listener(AppEvent::RESOURCE_RELEASE)
 */
class ResourceReleaseListener implements EventHandlerInterface
{
    /**
     * @param \Swoft\Event\EventInterface $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 \Swoft\Pool\ConnectionInterface $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);
        }
    }
}

考慮到用戶可能會在使用完後沒有釋放鏈接形成鏈接泄露，Swoft會在Rpc/Http請求或者Task結束後觸發一個Swoft.resourceRelease事件（注:Swoft是筆者添加的前綴，方便讀者區分Swoole相關事件和Swoft相關事件)，將鏈接強制收包並歸隊。