快捷導(dǎo)航

Swoole實(shí)現(xiàn)異步投遞task任務(wù)案例詳解

更新時(shí)間：2019年04月02日 11:00:49 作者：李維山

這篇文章主要介紹了Swoole實(shí)現(xiàn)異步投遞task任務(wù)案例,結(jié)合實(shí)例形式詳細(xì)分析了Swoole異步投遞task任務(wù)的原理、實(shí)現(xiàn)方法及操作注意事項(xiàng),需要的朋友可以參考下

本文實(shí)例講述了Swoole實(shí)現(xiàn)異步投遞task任務(wù)案例。分享給大家供大家參考，具體如下：

【使用場(chǎng)景】

Swolle的task模塊可以用來(lái)做一些異步的慢速任務(wù)、耗時(shí)場(chǎng)景。如webim中發(fā)廣播，發(fā)送郵件等，把這些任務(wù)丟給task進(jìn)程之后，worker進(jìn)程可以繼續(xù)處理新的數(shù)據(jù)請(qǐng)求，任務(wù)完成后會(huì)異步通知worker進(jìn)程告訴它此任務(wù)已經(jīng)完成。此外利用task還可以實(shí)現(xiàn)PHP的數(shù)據(jù)庫(kù)連接池，異步隊(duì)列等。

【使用須知】

必須設(shè)置Task進(jìn)程數(shù)： task_worker_num

投遞一個(gè)異步任務(wù)到task_worker池中：Server->task

必須注冊(cè)onTask、onFinish2個(gè)事件回調(diào)函數(shù)。

【示例】

以創(chuàng)建一個(gè)WebSocket服務(wù)異步投遞任務(wù)為例：

ws_server.php：

<?php

class Ws {
  public $ws = null;
  public function __construct() {
    $this->ws = new swoole_websocket_server("0.0.0.0", 9501);
    $this->ws->set([
        'worker_num' => 2, //worker進(jìn)程數(shù)
        'task_worker_num' => 2, //task進(jìn)程數(shù)
      ]);
    $this->ws->on("open", [$this, 'onOpen']);
    $this->ws->on("message", [$this, 'onMessage']);
    $this->ws->on("task", [$this, 'onTask']);
    $this->ws->on("finish", [$this, 'onFinish']);
    $this->ws->on("close", [$this, 'onClose']);
    $this->ws->start();
  }

  //建立連接回調(diào)
  public function onOpen($ws, $request) {
    echo "{$request->fd}建立了連接";
  }

  //接受消息回調(diào)
  public function onMessage($ws, $frame) {
    //worker進(jìn)程異步投遞任務(wù)到task_worker進(jìn)程中
    $data = [
      'fd' => $frame->fd,
    ];
    $ws->task($data);

    //服務(wù)器返回
    echo "服務(wù)器發(fā)送消息:666";
  }

  //完成異步任務(wù)回調(diào)
  public function onTask($serv, $task_id, $worker_id, $data) {
    var_dump($data);

    //模擬慢速任務(wù)
    sleep(5);

    //返回字符串給worker進(jìn)程——>觸發(fā)onFinish
    return "success";
  }

  //完成任務(wù)投遞回調(diào)
  public function onFinish($serv, $task_id, $data) {
    //task_worker進(jìn)程將任務(wù)處理結(jié)果發(fā)送給worker進(jìn)程
    echo "完成任務(wù){(diào)$task_id}投遞 處理結(jié)果：{$data}";
  }

  //關(guān)閉連接回調(diào)
  public function onClose($ws, $fd) {
    echo "{$fd}關(guān)閉了連接";
  }
}

$obj = new Ws();

前端頁(yè)面js監(jiān)聽(tīng)：

<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd">
<html xmlns="http://www.w3.org/1999/xhtml">
<head>
<title>WebSocket</title>
</head>
<body>
</body>
<script type="text/javascript">
var websocket = new WebSocket('ws://127.0.0.1:9501'); 

websocket.onopen = function (evt) { onOpen(evt) }; 
websocket.onclose = function (evt) { onClose(evt) }; 
websocket.onmessage = function (evt) { onMessage(evt) }; 
websocket.onerror = function (evt) { onError(evt) }; 
 
function onOpen(evt) {
  console.log("Connected to WebSocket server."); 
  
  //*發(fā)送消息到websocket服務(wù)器
  websocket.send('666');
} 
function onClose(evt) { 
  console.log("Disconnected"); 
}
function onMessage(evt) { 
  console.log('Retrieved data from server: ' + evt.data); 
} 
function onError(evt) { 
  console.log('Error occured: ' + evt.data); 
}
</script>
</html>

開(kāi)啟WebSocket服務(wù)：