如何在 Swoole 中優雅的實現 MySQL 鏈接池

如何在 Swoole 中優雅的實現 MySQL 鏈接池

1、爲何須要鏈接池 ？

數據庫鏈接池指的是程序和數據庫之間保持必定數量的鏈接不斷開，
而且各個請求的鏈接能夠相互複用，
減小重複鏈接數據庫帶來的資源消耗，
必定程度上提升了程序的併發性能。

2、鏈接池實現要點

• 協程：使用 MySQL 協程客戶端。

使用 MySQL 協程客戶端，是爲了能在一個 Worker 阻塞的時候，
讓出 CPU 時間片去處理其餘的請求，提升整個 Worker 的併發能力。

• 鏈接池存儲介質：使用 \swoole\coroutine\channel 通道。

使用 channel 可以設置等待時間，等待其餘的請求釋放鏈接。
而且在等待期間，一樣也可讓出 CPU 時間片去處理其餘的請求。

假設選擇 array 或 splqueue，沒法等待其餘的請求釋放鏈接。
那麼在高併發下的場景下，可能會出現鏈接池爲空的現象。
若是鏈接池爲空了，那麼 pop 就直接返回 null 了，致使鏈接不可用。

注：所以不建議選擇 array 或 splqueue。

3、鏈接池的具體實現

<?php

use Swoole\Coroutine\Channel;
use Swoole\Coroutine\MySQL;

class MysqlPool
{
    private $min; // 最小鏈接數
    private $max; // 最大鏈接數
    private $count; // 當前鏈接數
    private $connections; // 鏈接池
    protected $freeTime; // 用於空閒鏈接回收判斷

    public static $instance;

    /**
     * MysqlPool constructor.
     */
    public function __construct()
    {
        $this->min = 10;
        $this->max = 100;
        $this->freeTime = 10 * 3600;
        $this->connections = new Channel($this->max + 1);
    }

    /**
     * @return MysqlPool
     */
    public static function getInstance()
    {
        if (is_null(self::$instance)) {
            self::$instance = new self();
        }

        return self::$instance;
    }

    /**
     * 建立鏈接
     * @return MySQL
     */
    protected function createConnection()
    {
        $conn = new MySQL();
        $conn->connect([
            'host' => 'mysql',
            'port' => '3306',
            'user' => 'root',
            'password' => 'root',
            'database' => 'fastadmin',
            'timeout'  => 5
        ]);

        return $conn;
    }

    /**
     * 建立鏈接對象
     * @return array|null
     */
    protected function createConnObject()
    {
        $conn = $this->createConnection();
        return $conn ? ['last_used_time' => time(), 'conn' => $conn] : null;
    }

    /**
     * 初始化鏈接
     * @return $this
     */
    public function init()
    {
        for ($i = 0; $i < $this->min; $i++) {
            $obj = $this->createConnObject();
            $this->count++;
            $this->connections->push($obj);
        }

        return $this;
    }

    /**
     * 獲取鏈接
     * @param int $timeout
     * @return mixed
     */
    public function getConn($timeout = 3)
    {
        if ($this->connections->isEmpty()) {
            if ($this->count < $this->max) {
                $this->count++;
                $obj = $this->createConnObject();
            } else {
                $obj = $this->connections->pop($timeout);
            }
        } else {
            $obj = $this->connections->pop($timeout);
        }

        return $obj['conn']->connected ? $obj['conn'] : $this->getConn();
    }

    /**
     * 回收鏈接
     * @param $conn
     */
    public function recycle($conn)
    {
        if ($conn->connected) {
            $this->connections->push(['last_used_time' => time(), 'conn' => $conn]);
        }
    }

    /**
     * 回收空閒鏈接
     */
    public function recycleFreeConnection()
    {
        // 每 2 分鐘檢測一下空閒鏈接
        swoole_timer_tick(2 * 60 * 1000, function () {
           if ($this->connections->length() < intval($this->max * 0.5)) {
               // 請求鏈接數還比較多，暫時不回收空閒鏈接
               return;
           }

           while (true) {
               if ($this->connections->isEmpty()) {
                   break;
               }

               $connObj = $this->connections->pop(0.001);
               $nowTime = time();
               $lastUsedTime = $connObj['last_used_time'];

               // 當前鏈接數大於最小的鏈接數，而且回收掉空閒的鏈接
               if ($this->count > $this->min && ($nowTime - $lastUsedTime > $this->freeTime)) {
                   $connObj['conn']->close();
                   $this->count--;
               } else {
                   $this->connections->push($connObj);
               }
           }
        });
    }
}

$httpServer = new swoole_http_server('127.0.0.1',9501);
$httpServer->set(['work_num' => 1]);
$httpServer->on('WorkerStart', function ($request, $response) {
    MysqlPool::getInstance()->init()->recycleFreeConnection();
});
$httpServer->on('Request', function ($request, $response){
    $conn = MysqlPool::getInstance()->getConn();
    $conn->query('SELECT * FROM fa_admin WHERE id=1');
    MysqlPool::getInstance()->recycle($conn);
});
$httpServer->start();

4、總結

• 定時維護空閒鏈接到最小值。
• 使用用完數據庫鏈接以後，須要手動回收鏈接到鏈接池。
• 使用 channel 做爲鏈接池的存儲介質。