用swoole簡單實現MySQL鏈接池

MySQL鏈接池

在傳統的網站開發中，好比LNMP模式，由Nginx的master進程接收請求而後分給多個worker進程，每一個worker進程再連接php-fpm的master進程，php-fpm再根據當前狀況去調用其worker進程而後處理PHP，若是須要MySQL，在與MySQL創建鏈接，這個時候，若是有1000個請求打過來，就須要與MySQL創建1000個鏈接。若是請求達到萬級或者百萬級，這時數據庫的壓力就會很是大了。鏈接池技術就能夠派上用場了，能夠大大下降數據庫鏈接數，還能夠下降IO消耗。

爲何能夠下降IO消耗？

MySQL短鏈接每次請求操做數據庫都須要創建與MySQL服務器創建TCP鏈接，這是須要時間開銷的。TCP鏈接須要3次網絡通訊。這樣就增長了必定的延時和額外的IO消耗。請求結束後會關閉MySQL鏈接，還會發生3/4次網絡通訊。

而鏈接池採用的是MySQL的長鏈接模式，會一直保持與MySQL的鏈接，重用鏈接進行MySQL的操做，從而節省了創建鏈接和斷開鏈接的消耗。

爲何能夠下降數據鏈接數呢？

據的鏈接池爲維持若干個長鏈接，當新請求到達的時候，若是鏈接池空閒，就分配給鏈接池去處理，不然，後面的數據庫鏈接請求將被加入到等待隊列中。

用swoole實現一個簡單的MySQL鏈接池

<?php
$serv = new swoole_server('0.0.0.0', 9509);
$serv->set(array(
    'worker_num' => 50, //worker進程數量
    'task_worker_num' => 10, //task進程數量 即爲維持的MySQL鏈接的數量
));

function my_onReceive($serv, $fd, $from_id, $data)
{
    echo "收到數據".$data.PHP_EOL;
    //taskwait就是投遞一條任務，這裏直接傳遞SQL語句了
    //而後阻塞等待SQL完成，並返回結果
    $result = $serv->taskwait($data);
    echo "任務結束".PHP_EOL;
    if ($result !== false) {
        list($status, $db_res) = explode(':', $result, 2);
        if ($status == 'OK') {
            //數據庫操做成功了，執行業務邏輯代碼，這裏就自動釋放掉MySQL鏈接的佔用
            //將處理結果發送給客戶端
            $serv->send($fd, var_export(unserialize($db_res), true) . "\n");
        } else {
            $serv->send($fd, $db_res);
        }
        return;
    } else {
        $serv->send($fd, "Error. Task timeout\n");//若是返回的是false，則說明taskwait等待超時，能夠設置相應的等待超時時間
    }
}

function my_onTask($serv, $task_id, $from_id, $sql)
{
    echo "開始作任務 task id:".$task_id.PHP_EOL;
    static $link = null;
    HELL:
    if ($link == null) {
        $link = @mysqli_connect("127.0.0.1", "root", "passwd", "database");
        if (!$link) {
            $link = null;
            $serv->finish("ER:" . mysqli_error($link));
            return;
        }   
    }   
    $result = $link->query($sql);
    if (!$result) { //若是查詢失敗了
        if(in_array(mysqli_errno($link), [2013, 2006])){//錯誤碼爲2013，或者2006，則重連數據庫，從新執行sql
                $link = null;
                goto HELL;
        }else{
            $serv->finish("ER:" . mysqli_error($link));
            return;
        }
    }
    if(preg_match("/^select/i", $sql)){//若是是select操做，就返回關聯數組
        $data = $result->fetch_assoc();
    }else{//不然直接返回結果
        $data = $result;
    }
    $serv->finish("OK:" . serialize($data));//調用finish方法，用於在task進程中通知worker進程，投遞的任務已完成
    //return "OK:".serialize($data);
}

function my_onFinish($serv, $task_id, $data)
{
    echo "任務完成";//taskwait 沒有觸發這個函數。。
    echo "AsyncTask Finish:Connect.PID=" . posix_getpid() . PHP_EOL;
}

$serv->on('receive', 'my_onReceive');
$serv->on('task', 'my_onTask');
$serv->on('Finish', 'my_onFinish');

$serv->start();//啓動server