PHP開發異步高性能的MySQL代理服務器

MySQL數據庫對每個客戶端連接都會分配一個線程，所以連接非常寶貴。開發一個異步的MySQL代理服務器，PHP應用服務器可以長連接到這臺Server，既減輕MYSQL的連接壓力，又使PHP保持長連接減少connect/close的網絡開銷。

此Server考慮到了設置了數據庫連接池尺寸，區分忙閑，mysqli斷線重連，并設置了負載保護。基于swoole擴展開發，io循環使用epoll，是全異步非阻塞的，可以應對大量TCP連接。

程序的邏輯是：啟動時創建N個MySQL連接，收到客戶端發來的SQL后，分配1個MySQL連接，將SQL發往數據庫服務器。然后等待數據庫返回查詢結果。當數據庫返回結果后，再發給對應的客戶端連接。

核心的數據結構是3個PHP數組。idle_pool是空閑的數據庫連接，當有SQL請求時從idle_pool中移到busy_pool中。當數據庫返回結果后從busy_pool中再移到idle_pool中，以供新的請求使用。當SQL請求到達時如果沒有空閑的數據庫連接，那會自動加入到wait_queue中。一旦有SQL完成操作，將自動從wait_queue中取出等待的請求進行處理。

如此循環使用。由于整個服務器是異步的單進程單線程所以完全不需要鎖。而且是完全異步的，效率非常高。

當然本文的代碼，如果要用于生產環境，還需做更多的保護機制和壓力測試。在此僅拋磚引玉，提供一個解決問題的思路。

class DBServer
{
    protected $pool_size = 20;
    protected $idle_pool = array(); //空閑連接
    protected $busy_pool = array(); //工作連接
    protected $wait_queue = array(); //等待的請求
    protected $wait_queue_max = 100; //等待隊列的最大長度，超過后將拒絕新的請求

    /**
     * @var swoole_server
     */
    protected $serv;

    function run()
    {
        $serv = new swoole_server("127.0.0.1", 9509);
        $serv->set(array(
            'worker_num' => 1,
        ));

        $serv->on('WorkerStart', array($this, 'onStart'));
        //$serv->on('Connect', array($this, 'onConnect'));
        $serv->on('Receive', array($this, 'onReceive'));
        //$serv->on('Close', array($this, 'onClose'));
        $serv->start();
    }

    function onStart($serv)
    {
        $this->serv = $serv;
        for ($i = 0; $i < $this->pool_size; $i++) {
            $db = new mysqli;
            $db->connect('127.0.0.1', 'root', 'root', 'test');
            $db_sock = swoole_get_mysqli_sock($db);
            swoole_event_add($db_sock, array($this, 'onSQLReady'));
            $this->idle_pool[] = array(
                'mysqli' => $db,
                'db_sock' => $db_sock,
                'fd' => 0,
            );
        }
        echo "Server: start.Swoole version is [" . SWOOLE_VERSION . "]\n";
    }

    function onSQLReady($db_sock)
    {
        $db_res = $this->busy_pool[$db_sock];
        $mysqli = $db_res['mysqli'];
        $fd = $db_res['fd'];

        echo __METHOD__ . ": client_sock=$fd|db_sock=$db_sock\n";

        if ($result = $mysqli->reap_async_query()) {
            $ret = var_export($result->fetch_all(MYSQLI_ASSOC), true) . "\n";
            $this->serv->send($fd, $ret);
            if (is_object($result)) {
                mysqli_free_result($result);
            }
        } else {
            $this->serv->send($fd, sprintf("MySQLi Error: %s\n", mysqli_error($mysqli)));
        }
        //release mysqli object
        $this->idle_pool[] = $db_res;
        unset($this->busy_pool[$db_sock]);

        //這里可以取出一個等待請求
        if (count($this->wait_queue) > 0) {
            $idle_n = count($this->idle_pool);
            for ($i = 0; $i < $idle_n; $i++) {
                $req = array_shift($this->wait_queue);
                $this->doQuery($req['fd'], $req['sql']);
            }
        }
    }

    function onReceive($serv, $fd, $from_id, $data)
    {
        //沒有空閑的數據庫連接
        if (count($this->idle_pool) == 0) {
            //等待隊列未滿
            if (count($this->wait_queue) < $this->wait_queue_max) {
                $this->wait_queue[] = array(
                    'fd' => $fd,
                    'sql' => $data,
                );
            } else {
                $this->serv->send($fd, "request too many, Please try again later.");
            }
        } else {
            $this->doQuery($fd, $data);
        }
    }

    function doQuery($fd, $sql)
    {
        //從空閑池中移除
        $db = array_pop($this->idle_pool);
        /**
         * @var mysqli
         */
        $mysqli = $db['mysqli'];

        for ($i = 0; $i < 2; $i++) {
            $result = $mysqli->query($sql, MYSQLI_ASYNC);
            if ($result === false) {
                if ($mysqli->errno == 2013 or $mysqli->errno == 2006) {
                    $mysqli->close();
                    $r = $mysqli->connect();
                    if ($r === true) continue;
                }
            }
            break;
        }

        $db['fd'] = $fd;
        //加入工作池中
        $this->busy_pool[$db['db_sock']] = $db;
    }
}

$server = new DBServer();
$server->run();