用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