Swoft 源码剖析 - 连接池

作者：bromine

链接： https://www.jianshu.com/p/1a7...

來源：简书

著作权归作者所有，本文已获得作者授权转载，并对原文进行了重新的排版。

Swoft Github: https://github.com/swoft-clou...

为什么需要引入连接池?

对于基于php-fpm的传统php-web应用，包括且不限于Mysql,Redis,RabbitMq，每次请求到来都需要为其新建一套独享的的连接，这直接带来了一些典型问题：

1. 连接开销 :连接随着http请求到来而新建,随着请求返回而销毁，大量连接新建销毁是对系统资源的浪费。
2. 连接数量过高 :每一个请求都需要一套自己的连接，系统连接数和并发数会成一个近线性的关系。如果系统并发量达到了1w,那么就需要建立1w个对应的连接，这对于Mysql之类的后端服务而言，是一个大的负荷。
3. 空闲连接 :假设我们有一个接口使用了一个Mysql连接。该接口在一开始进行一次sql查询后，后面的操作都是sql无关的，那么该请求占据的空闲连接完全就是一种资源的浪费。

对于异步系统而言，这个问题变得更加的严峻。一个请求处理进程要对同一个服务进行并发的操作，意味着这个请求要持有1个以上同类的连接，这对于系统压力而言，无疑是雪上加霜了，所以连接池对于基于Swoole的Web框架而言已经是一个必需实现的机制了。

Swoft连接池的生命周期与进程模型

连接池作为一个 SCOPE 为 SINGLETON 的典型 Bean ,

其实例最早会在 Swoft\Bean\BeanFactory::reload() 阶段被初始化。

Worker/Task进程

对于RPC或者HTTP请求而言，关系最密切的进程肯定是Worker和Task进程了。

对于这两者而言 SwoftBeanBeanFactory::reload()会在swoole的onWorkerStart事件的回调阶段阶段被调用。

//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个

1. 多个进程同时对一个连接进行读写会导致数据传输错乱 ，需要保证连接不会被同时访问。
2. Worker进程对程序全局期的对象进行写操作时会导致写时复制，产生一个进程全局期的副本，程序全局期较难维持。
3. 使用进程全局期的话可以利用现有的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() 启动 。

但是无论是何者，最终都会在Process中调用 beforeProcess() 进行子进程的初始化。对于 boot为true的 前置进程 ，由于其启动时父进程还未初始化bean容器，所以会单独进行bean容器初始化，而对于boot为false的其他 用户自定义进程 ,其会直接继承父进程的Ioc容器。

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相似且更简单，严格遵循一个进程一个连接池，这里不再累述。

假设Worker数目为j,Task数目为k,Process数为l,Command数为m，每个进程池内配置最大连接数为n，部署机器数为x，不难看出每个swoft项目占用的连接数为 (j+k+l+m)*n*x 。

这种方案中，项目占用的连接数仅仅为 k*x 。

除了Task进程各个进程并不直接持有连接池，而是通过向Task进程提交指令( task() , sendMessage() )让其代为进行连接池相关服务的操作，至少需要额外的一次进程间通信（默认为Unix Socket）。

该方案虽然能够更好的复用连接和节省连接数，但机制实现并不方便。从另一个角度去看，Swoft的连接池方案是为了解决使用Swoole时，单进程并发执行的连接数要求问题；Range提出的连接池方案是为了解决超大流量系统下对Mysql等服务的压力控制问题。两者适合不同的场景，其目的和意义在一定程度下是重合的，但并不是完全一样的。

Swoft连接池的实现

池的容器

连接池根据当前是否协程环境选择一种合适的队列结构作为连接的容器。

\SplQueue:SplQueue
\Swoole\Coroutine\Channel

连接的获取

\\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相关事件)，将连接强制收包并归队。

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