摘要：要實現(xiàn)上述的異步處理，只需要增加兩個事件回調(diào)即可和這兩個回調(diào)函數(shù)分別用于執(zhí)行任務(wù)和處理任務(wù)的返回結(jié)果。還需要注意的是，服務(wù)器如果檢測到長時間沒有沒有查詢，則會斷開連接回收資源所以要有斷線重連的機制。

在一般的 Server 程序中都會有一些耗時的任務(wù)，比如：發(fā)送郵件、聊天服務(wù)器發(fā)送廣播等。如果我們采用同步阻塞的防水去執(zhí)行這些任務(wù)，那么這肯定會非常的慢。

Swoole 的 TaskWorker 進程池可以用來執(zhí)行一些異步的任務(wù)，而且不會影響接下來的任務(wù)，很適合處理以上場景。

那么什么是異步任務(wù)呢？

可以從下面的圖示中來簡單了解一下。（來源于網(wǎng)絡(luò)，侵刪）

我們上一個 Swoole 的文章介紹了如何創(chuàng)建一個簡單的服務(wù)器，并且知道了幾個核心的回調(diào)函數(shù)的使用方法。

要實現(xiàn)上述的異步處理，只需要增加兩個事件回調(diào)即可：onTask 和 onFinish, 這兩個回調(diào)函數(shù)分別用于執(zhí)行 Task 任務(wù)和處理 Task 任務(wù)的返回結(jié)果。另外還需要在 set 方法中設(shè)置 task 進程數(shù)量。

使用示例：

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}
";
        // 發(fā)送任務(wù)到Task進程
        $param = array(
            "fd" => $fd
        );
        $serv->task( json_encode( $param ) );
        echo "繼續(xù)處理之后的邏輯
";
    }

    public function onTask($serv, $task_id, $from_id, $data) {
        echo "This Task {$task_id} from Worker {$from_id}
";
        echo "Data: {$data}
";
        for($i = 0 ; $i < 5 ; $i ++ ) {
            sleep(1);
            echo "Task {$task_id} Handle {$i} times...
";
        }
        $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
";
        echo "Result: {$data}
";
    }
    public function onStart( $serv ) {
        echo "Server Start
";
    }
    public function onConnect( $serv, $fd, $from_id ) {
        echo "Client {$fd} connect
";
    }
    public function onClose( $serv, $fd, $from_id ) {
        echo "Client {$fd} close connection
";
    }
}
$server = new Server();

通過上述示例可以看到，發(fā)起一個異步任務(wù)只需要調(diào)用 swoole_server 的 task 方法就可以。發(fā)送之后會觸發(fā) onTask 回調(diào)，可以通過 $task_id 和 $from_id 處理不同進程的不同任務(wù)。最后可以通過 return 一個字符串來將執(zhí)行結(jié)果返回給 Worker 進程，Worker 進程通過 onFinish 回調(diào)來處理結(jié)果。

那么基于上述代碼就可以實現(xiàn)異步操作 mysql。異步操作 mysql 較適合以下場景：

并發(fā)的讀寫操作

沒有時序上的嚴格關(guān)系

不影響主線程邏輯

好處：

提高并發(fā)

降低 IO 消耗

數(shù)據(jù)庫的壓力主要在于 mysql 維持的連接數(shù)，如果存在 1000 個并發(fā)，那么 mysql 就需要建立對應(yīng)數(shù)量的連接。而采用長連接的方式，mysql 的連接一直維持在進程中，減少了創(chuàng)建連接的損耗?？梢酝ㄟ^ swoole 開啟多個 task 進程，每一個進程內(nèi)維持一個mysql 長連接，那么這樣子也可以引申出來 mysql 連接池技術(shù)。還需要注意的是，mysql 服務(wù)器如果檢測到長時間沒有沒有查詢，則會斷開連接回收資源,所以要有斷線重連的機制。

以下是一個簡單的異步操作 mysql 的示例：

還是以上的代碼，我們只需要修改 onReceive、onTask、onFinish 三個函數(shù)。

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進程數(shù)量 即為維持的MySQL連接的數(shù)量
        ));
        $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 "收到數(shù)據(jù)". $data . PHP_EOL;
        // 發(fā)送任務(wù)到Task進程
        $param = array(
            "sql" => $data, // 接收客戶端發(fā)送的 sql 
            "fd"  => $fd
        );
        $serv->task( json_encode( $param ) );  // 向 task 投遞任務(wù)
        echo "繼續(xù)處理之后的邏輯
";
    }

    public function onTask($serv, $task_id, $from_id, $data) {
        echo "This Task {$task_id} from Worker {$from_id}
";
        echo "recv SQL: {$data["sql"]}
";
        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，則重連數(shù)據(jù)庫，重新執(zhí)行sql
                    $link = null;
                    goto HELL;
            }
        }
        if(preg_match("/^select/i", $sql)){//如果是select操作，就返回關(guān)聯(lián)數(shù)組
             $data = array();
                while ($fetchResult = mysqli_fetch_assoc($result) ){
                     $data["data"][] = $fetchResult;
                }                
        }else{//否則直接返回結(jié)果
            $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
";
        $result = json_decode($result, true);
        if ($result["status"] == "OK") {
            $this->serv->send($result["fd"], json_encode($result["data"]) . "
");
        } else {
            $this->serv->send($result["fd"], $result);
        }
    }
    public function onStart( $serv ) {
        echo "Server Start
";
    }
    public function onConnect( $serv, $fd, $from_id ) {
        echo "Client {$fd} connect
";
    }
    public function onClose( $serv, $fd, $from_id ) {
        echo "Client {$fd} close connection
";
    }
}
$server = new Server();

以上代碼在 onReceive 時直接接收一條 sql，之后直接發(fā)送到 Task 任務(wù)中。這個時候下一步的流程緊接著輸出，這里也就體現(xiàn)出了異步。然后 onTask 和 onFinish 分別用來向數(shù)據(jù)庫發(fā)送 sql，處理 task 執(zhí)行結(jié)果。

參考鏈接：

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