PHP 使用 Swoole - TaskWorker 实现异步操做 Mysql

在通常的 Server 程序中都会有一些耗时的任务，好比：发送邮件、聊天服务器发送广播等。若是咱们采用同步阻塞的防水去执行这些任务，那么这确定会很是的慢。

Swoole 的 TaskWorker 进程池能够用来执行一些异步的任务，并且不会影响接下来的任务，很适合处理以上场景。

那么什么是异步任务呢？

能够从下面的图示中来简单了解一下。（来源于网络，侵删）

咱们上一个 Swoole 的文章介绍了如何建立一个简单的服务器，而且知道了几个核心的回调函数的使用方法。

要实现上述的异步处理，只须要增长两个事件回调便可：onTask 和 onFinish, 这两个回调函数分别用于执行 Task 任务和处理 Task 任务的返回结果。另外还须要在 set 方法中设置 task 进程数量。

使用示例：

class Server
{
    private $serv;
    public function __construct() {
        $this->serv = new swoole_server("0.0.0.0", 9501);
        $this->serv->set(array(
            'worker_num' => 4,
            'daemonize' => false,
            'task_worker_num' => 8
        ));
        $this->serv->on('Start', array($this, 'onStart'));
        $this->serv->on('Connect', array($this, 'onConnect'));
        $this->serv->on('Receive', array($this, 'onReceive'));
        $this->serv->on('Close', array($this, 'onClose'));
        $this->serv->on('Task', array($this, 'onTask'));
        $this->serv->on('Finish', array($this, 'onFinish'));
        $this->serv->start();
    }

    public function onReceive( swoole_server $serv, $fd, $from_id, $data ) {
        echo "Get Message From Client {$fd}:{$data}\n";
        // 发送任务到Task进程
        $param = array(
            'fd' => $fd
        );
        $serv->task( json_encode( $param ) );
        echo "继续处理以后的逻辑\n";
    }

    public function onTask($serv, $task_id, $from_id, $data) {
        echo "This Task {$task_id} from Worker {$from_id}\n";
        echo "Data: {$data}\n";
        for($i = 0 ; $i < 5 ; $i ++ ) {
            sleep(1);
            echo "Task {$task_id} Handle {$i} times...\n";
        }
        $fd = json_decode( $data , true )['fd'];
        $serv->send( $fd , "Data in Task {$task_id}");
        return "Task {$task_id}'s result";
    }
    public function onFinish($serv,$task_id, $data) {
        echo "Task {$task_id} finish\n";
        echo "Result: {$data}\n";
    }
    public function onStart( $serv ) {
        echo "Server Start\n";
    }
    public function onConnect( $serv, $fd, $from_id ) {
        echo "Client {$fd} connect\n";
    }
    public function onClose( $serv, $fd, $from_id ) {
        echo "Client {$fd} close connection\n";
    }
}
$server = new Server();

经过上述示例能够看到，发起一个异步任务只须要调用 swoole_server 的 task 方法就能够。发送以后会触发 onTask 回调，能够经过 $task_id 和 $from_id 处理不一样进程的不一样任务。最后能够经过 return 一个字符串来将执行结果返回给 Worker 进程，Worker 进程经过 onFinish 回调来处理结果。

那么基于上述代码就能够实现异步操做 mysql。异步操做 mysql 较适合如下场景：

• 并发的读写操做
• 没有时序上的严格关系
• 不影响主线程逻辑

好处：

• 提升并发
• 下降 IO 消耗

数据库的压力主要在于 mysql 维持的链接数，若是存在 1000 个并发，那么 mysql 就须要创建对应数量的链接。而采用长链接的方式，mysql 的链接一直维持在进程中，减小了建立链接的损耗。能够经过 swoole 开启多个 task 进程，每个进程内维持一个mysql 长链接，那么这样子也能够引伸出来 mysql 链接池技术。还须要注意的是，mysql 服务器若是检测到长时间没有没有查询，则会断开链接回收资源,因此要有断线重连的机制。

如下是一个简单的异步操做 mysql 的示例：

仍是以上的代码，咱们只须要修改 onReceive、onTask、onFinish 三个函数。

class Server
{
    private $serv;
    public function __construct() {
        $this->serv = new swoole_server("0.0.0.0", 9501);
        $this->serv->set(array(
            'worker_num' => 4,
            'daemonize' => false,
            'task_worker_num' => 8 // task进程数量 即为维持的MySQL链接的数量
        ));
        $this->serv->on('Start', array($this, 'onStart'));
        $this->serv->on('Connect', array($this, 'onConnect'));
        $this->serv->on('Receive', array($this, 'onReceive'));
        $this->serv->on('Close', array($this, 'onClose'));
        $this->serv->on('Task', array($this, 'onTask'));
        $this->serv->on('Finish', array($this, 'onFinish'));
        $this->serv->start();
    }

    public function onReceive( swoole_server $serv, $fd, $from_id, $data ) {
        echo "收到数据". $data . PHP_EOL;
        // 发送任务到Task进程
        $param = array(
            'sql' => $data, // 接收客户端发送的 sql 
            'fd'  => $fd
        );
        $serv->task( json_encode( $param ) );  // 向 task 投递任务
        echo "继续处理以后的逻辑\n";
    }

    public function onTask($serv, $task_id, $from_id, $data) {
        echo "This Task {$task_id} from Worker {$from_id}\n";
        echo "recv SQL: {$data['sql']}\n";
        static $link = null;
        $sql = $data['sql'];
        $fd  = $data['fd'];
        HELL:
        if ($link == null) {
            $link = @mysqli_connect("127.0.0.1", "root", "root", "test");
        }
        $result = $link->query($sql);
        if (!$result) { //若是查询失败
            if(in_array(mysqli_errno($link), [2013, 2006])){
                //错误码为2013，或者2006，则重连数据库，从新执行sql
                    $link = null;
                    goto HELL;
            }
        }
        if(preg_match("/^select/i", $sql)){//若是是select操做，就返回关联数组
             $data = array();
                while ($fetchResult = mysqli_fetch_assoc($result) ){
                     $data['data'][] = $fetchResult;
                }                
        }else{//不然直接返回结果
            $data['data'] = $result;
        }
        $data['status'] = "OK";
        $data['fd'] = $fd;
        $serv->finish(json_encode($data));
    }
    public function onFinish($serv, $task_id, $data) {
        echo "Task {$task_id} finish\n";
        $result = json_decode($result, true);
        if ($result['status'] == 'OK') {
            $this->serv->send($result['fd'], json_encode($result['data']) . "\n");
        } else {
            $this->serv->send($result['fd'], $result);
        }
    }
    public function onStart( $serv ) {
        echo "Server Start\n";
    }
    public function onConnect( $serv, $fd, $from_id ) {
        echo "Client {$fd} connect\n";
    }
    public function onClose( $serv, $fd, $from_id ) {
        echo "Client {$fd} close connection\n";
    }
}
$server = new Server();

以上代码在 onReceive 时直接接收一条 sql，以后直接发送到 Task 任务中。这个时候下一步的流程紧接着输出，这里也就体现出了异步。而后 onTask 和 onFinish 分别用来向数据库发送 sql，处理 task 执行结果。

参考连接：

https://wiki.swoole.com
http://rango.swoole.com/archi...