swoole-framwork/core/lib/DBPool.php

<?php
namespace Core\lib;

abstract class DBPool
{
    private $min;
    private $max;
    private $conns;
    private $count;
    private $idleTime = 40;
    abstract protected function newDB();

    function __construct($min = 5, $max = 10)
    {
        $this->min = $min;
        $this->max = $max;
        $this->conns = new \Swoole\Coroutine\Channel($this->max);
        for ($i = 0; $i < $min; $i++) {
            $this->addDBToPool();
        }

        \Swoole\Timer::tick(30000, [$this, 'keepAlive']);
        //\Swoole\Timer::tick(30000, [$this, 'keepAlive']);
    }

    public function keepAlive()
    {
        for ($i = 0; $i < $this->conns->length(); $i++) {
            $dbObj = $this->getConnection();
            try {
                $res = $dbObj->db->query('select 1');
                //var_dump($res);
                $this->close($dbObj);
            }catch (\Exception $e) {
                throw new \Exception($e->getMessage());
            }
        }


    }

    public function initPool()
    {
        /*for ($i = 0; $i < $this->min; $i++) {
            $db = $this->newDB();
            $this->conns->push($db);
        }
        $this->count = $this->min;*/
    }

    public function getConnection()
    {
        if($this->conns->isEmpty()){
            if ($this->count < $this->max) {
                $this->addDBToPool();
                $dbObject = $this->conns->pop();
            }else{
                $dbObject = $this->conns->pop(5);
            }
        }else{
            $dbObject = $this->conns->pop();
        }
        if ($dbObject) {
            $dbObject->usedTime = time();
        }
        /**$res = $this->pdoPing($dbObject->db);
        if (!$res) {
            $this->count--;
            $this->addDBToPool();
            return $this->getConnection();
        }*/
        return $dbObject;
    }

    function pdoPing($dbconn){
        try{
            $dbconn->getAttribute(\PDO::ATTR_SERVER_INFO);
        } catch (\Exception $e) {
            /*if(strpos($e->getMessage(), 'MySQL server has gone away')!==false){
                return false;
            }*/
            return false;
        }
        return true;
    }

    public function close($conn)
    {
        if ($conn) {
            $this->conns->push($conn);
        }
    }

    public function addDBToPool()
    {
        try {
            $this->count++;
            $db = $this->newDB();
            if (!$db) throw new \Exception('db创建错误');
            $dbObject = new \stdClass();
            $dbObject->usedTime = time();
            $dbObject->db = $db;
            $this->conns->push($dbObject);
        }catch (\Exception $e) {
            $this->count--;
        }
    }

    public function cleanPool()
    {
        if ($this->conns->length() < $this->min && $this->conns->length()<intval($this->max * 0.6)) {
            return ;
        }
        $dbbak = [];
        while (true) {
            if($this->conns->isEmpty()) {
                break;
            }
            $obj = $this->conns->pop(0.1);
            if($this->count > $this->min && (time()-$obj->usedTime) > $this->idleTime){
                $this->count--;
            } else{
                $dbbak[] = $obj;
            }
        }
        foreach ($dbbak as $db){
            $this->conns->push($db);
        }
    }
}