Swoole异步投递task任务

青旅半醒 2023-10-06 17:32 118阅读 0赞

#### 阅读目录 ####

*  Swoole异步投递task任务
 *   *  【使用场景】
     *  【使用须知】
     *  【示例】
     *   *  ws\_server.php
         *  index.html
 *  php swoole4 执行异步任务(Task)
 *   *  WS.php写服务端代码
     *  chat.html写客户端代码
     *  测试效果
     *   *  1、分别打开服务端和客户端
         *  2、发送信息测试
         *  3、等待10秒后

## Swoole异步投递task任务 ##

实现实例前提：需要安装swoole  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaWd1YW5nMTAy_size_16_color_FFFFFF_t_70]

### 【使用场景】 ###

Swoole的 task 模块可以用来做一些异步的慢速任务、耗时场景。

如webim中发广播，发送邮件等，把这些任务丢给task进程之后，worker进程可以继续处理新的数据请求，任务完成后会异步通知worker进程告诉它此任务已经完成。

此外利用 task 还可以实现PHP的数据库连接池，异步队列等。

### 【使用须知】 ###

必须设置Task进程数：task\_worker\_num

投递一个异步任务到task\_worker池中：Server->task

必须注册onTask、onFinish2个事件回调函数。

### 【示例】 ###

以创建一个WebSocket服务异步投递任务为例：

#### 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进程数
            'task_worker_num' => 2, //task进程数
          ]);
        $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();
      }
    
      //建立连接回调
      public function onOpen($ws, $request) {
        
        echo "{
          $request->fd}建立了连接".PHP_EOL;
      }
    
      //接受消息回调
      public function onMessage($ws, $frame) {
        
        //worker进程异步投递任务到task_worker进程中
        $data = [
          'fd' => $frame->fd,
        ];
        $ws->task($data);
    
        //服务器返回
        echo "服务器发送消息:666".PHP_EOL;
      }
    
      //完成异步任务回调
      public function onTask($serv, $task_id, $worker_id, $data) {
        
        var_dump($data);
    
        //模拟慢速任务
        sleep(5);
    
        //返回字符串给worker进程——>触发onFinish
        return "success";
      }
    
      //完成任务投递回调
      public function onFinish($serv, $task_id, $data) {
        
        //task_worker进程将任务处理结果发送给worker进程
        echo "完成任务{
          $task_id}投递 处理结果：{
          $data}";
      }
    
      //关闭连接回调
      public function onClose($ws, $fd) {
        
        echo "{
          $fd}关闭了连接";
      }
    }
    
    $obj = new Ws();

#### index.html ####

<!DOCTYPE html>
    <head>
    <title>WebSocket</title>
    </head>
    <body>
    </body>
    <script type="text/javascript">
    var websocket = new WebSocket('ws://test.cc: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."); 
    
      //*发送消息到websocket服务器
      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>

## php swoole4 执行异步任务(Task) ##

异步任务：  
Swoole 提供了异步任务处理的功能，可以投递一个异步任务到 TaskWorker 进程池中执行，不影响当前请求的处理速度。

场景：  
在 Server 程序中如果需要执行很耗时的操作，比如一个聊天服务器发送广播，Web 服务器中发送邮件。如果直接去执行这些函数就会阻塞当前进程，导致服务器响应变慢。

### WS.php写服务端代码 ###

<?php
    class WS{
        
        private $ws = null;
        
        public function __construct(){
        
            //创建WebSocket Server对象，监听0.0.0.0:9502端口
            $this->ws = new Swoole\WebSocket\Server('0.0.0.0', 9502);
            $this->ws->set([
                'task_worker_num' => 4//设置异步任务的工作进程数量
            ]);
            //监听WebSocket连接打开事件
            $this->ws->on('Open', [$this, "onOpen"]);
            //监听WebSocket消息事件
            $this->ws->on('Message', [$this, "onMessage"]);
            
            //处理异步任务(此回调函数在task进程中执行)
            $this->ws->on('Task', [$this, "onTask"]);
            //处理异步任务的结果(此回调函数在worker进程中执行)
            $this->ws->on('Finish', [$this, "onFinish"]);
            
            //监听WebSocket连接关闭事件
           $this->ws->on('Close', [$this, "onClose"]);
            //启动服务器
           $this->ws->start();
        }
        public function onOpen($ws, $request){
        
            //var_dump($request->fd, $request->get, $request->server);
            $ws->push($request->fd, "欢迎客户端： {
          $request->fd}\n");
        }
        public function onMessage($ws, $frame){
        
            echo "信息: {
          $frame->data}\n";
            foreach($ws->connections as $fd){
        
                //$fd指单前用户，$frame->fd外来连接的用户
                if($fd == $frame->fd){
        
                	
                	/*投递异步任务,执行完会调onTask,整个程序不会等待task任务执行完，
                	而会直接同步执行下面的代码，提前输出信息
                	*/
                    $ws->task([
                        'fd' => $fd,
                        'message' => "我: {
          $frame->data}"
                    ]);
                    
                    $ws->push($fd, "我: {
          $frame->data}");
                }else{
        
                    $ws->push($fd, "对方: {
          $frame->data}");
                }
            }
        }
        public function onTask($ws, $task_id, $reactor_id, $data){
        
            sleep(10);//为了能看到对比效果，执行休眠10秒
            return $data;//return到onFinish
        }
        
        public function onFinish($ws, $task_id, $data){
        
            var_dump("task_id：". $task_id);
            var_dump($data);
        }
        public function onClose($ws, $fd){
        
            echo "客服端：{
          $fd} 关闭\n";
        }
    }
    new WS();

### chat.html写客户端代码 ###

<!doctype html>
    <html>
    <head>
    <meta charset="utf-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1, user-scalable=no">
    <title>简易聊天室</title>
    </head>
    <div id="welcome"></div>
    <input type="text" id="input"/>
    <input type="buton" onclick="send()" value="发送"/>
    <div id="message"></div>
    <body>
    </body>
    <script src="https://code.jquery.com/jquery-3.1.1.min.js"></script>
    <script>
        var wsServer = 'ws://test.cc:9502';
        var websocket = new WebSocket(wsServer);
        websocket.onopen = function (res) {
        
            $("#welcome").append("<h1>连接成功！欢迎</h1>");
        };
        websocket.onclose = function (res) {
        
            $("#message").append("<h3>连接关闭</h3>");
        };
        websocket.onmessage = function (res) {
        
            $("#message").append("<h3>" + res.data + "</h3>");
        };
        websocket.onerror = function (res) {
        
            $("#message").append("<h3>" + res + "</h3>");
        };
        function send(){
        
            websocket.send($('#input').val());
        }
    </script>
    </html>

### 测试效果 ###

#### 1、分别打开服务端和客户端 ####

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2djaGVufg_size_20_color_FFFFFF_t_70_g_se_x_16]

#### 2、发送信息测试 ####

结果：刚开始执行发送信息，虽然task任务被我sleep了10秒，但是task前后的打印都直接执行了  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2djaGVufg_size_20_color_FFFFFF_t_70_g_se_x_16 1]

#### 3、等待10秒后 ####

结果：task任务的打印出来了

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2djaGVufg_size_17_color_FFFFFF_t_70_g_se_x_16]  
结果说明  
开启task异步任务后，整个程序不会等待task任务执行完，而会直接同步执行下面的代码，提前输出信息，及task里的代码不会影响整个程序的运行速度。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaWd1YW5nMTAy_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20210430114152260.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaWd1YW5nMTAy,size_16,color_FFFFFF,t_70
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2djaGVufg_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/c75e90100c8c45df91148cda781fba7e.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAd2djaGVufg==,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2djaGVufg_size_20_color_FFFFFF_t_70_g_se_x_16 1]: https://img-blog.csdnimg.cn/a7e2a634744e41d3820832975885e081.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAd2djaGVufg==,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2djaGVufg_size_17_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/cfd400a67c3b458bad619d2779233c84.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAd2djaGVufg==,size_17,color_FFFFFF,t_70,g_se,x_16