PHP 使用 Swoole - TaskWorker 實現異步操做 Mysql

在通常的 Server 程序中都會有一些耗時的任務,好比:發送郵件、聊天服務器發送廣播等。若是咱們採用同步阻塞的防水去執行這些任務,那麼這確定會很是的慢。mysql

Swoole 的 TaskWorker 進程池能夠用來執行一些異步的任務,並且不會影響接下來的任務,很適合處理以上場景。sql

那麼什麼是異步任務呢?數據庫

能夠從下面的圖示中來簡單瞭解一下。(來源於網絡,侵刪)
json

咱們上一個 Swoole 的文章介紹瞭如何建立一個簡單的服務器,而且知道了幾個核心的回調函數的使用方法。數組

要實現上述的異步處理,只須要增長兩個事件回調便可:onTask 和 onFinish, 這兩個回調函數分別用於執行 Task 任務和處理 Task 任務的返回結果。另外還須要在 set 方法中設置 task 進程數量。 服務器

使用示例:swoole

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...

相關文章
相關標籤/搜索