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

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

    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();

通過上述示例可以看到，發起一個異步任務只需要調用 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 "繼續處理之后的邏輯
";
    }

    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，則重連數據庫，重新執行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
";
        $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，之后直接發送到 Task 任務中。這個時候下一步的流程緊接著輸出，這里也就體現出了異步。然后 onTask 和 onFinish 分別用來向數據庫發送 sql，處理 task 執行結果。

參考鏈接：

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