thinkphp-queue 消息队列服务

秒速五厘米 2022-12-06 04:20 331阅读 0赞

**[thinkphp-queue][]** 是thinkphp 官方提供的一个消息队列服务，它支持消息队列的一些基本特性：

*  消息的**发布**，**获取**，**执行**，**删除**，**重发**，**失败处理**，**延迟执行**，**超时控制**等
 *  队列的**多队列**， **内存限制** ，**启动**，**停止**，**守护**等
 *  消息队列可**降级为同步执行**
 *

搭建消息队列的存储环境

*  可使用 Redis \[**推荐**\]
 *  可使用数据库

**推荐使用redis**

首先安装redis

这次使用的是宝塔，安装流程：

### 一、[宝塔Linux面板安装Redis][Linux_Redis] ###

宝塔Linux面板安装Redis不会特别麻烦，只要几步就可以实现：

1.安装redis服务

2.配置redis设置【可不设置】

3.安装PHP扩展

1.首先，我们来安装redis服务，进入宝塔管理面板--软件管理--运行环境--redis-点击安装，等待完成

![1f3c50a1d4c0ee865311bd239625dfa7.png][]

2.完成之后开始第二步，配置redis设置。这一步根据自己需要进行配置。注意安全问题哦

我这里就改一下设置密码验证

找到/www/server/redis/下面的文件redis.conf里面的requirepass foobared这行，将注释去掉

![0fc71b3f36025c52bd144fb22b300eb6.png][]

foobared这个改为自己想设置的密码

3.然后安装php扩展

点击PHP对应的设置--安装扩展--redis--安装

![18dba49f11b5a68880c79ac0d4fb575f.png][]

等待完成。

完成后重启即可。

**注意**：1.需要开启相应的端口，宝塔面板-安全，还有服务器提供商那边也开放此端口

2.若宝塔上安装了多个版本的php，可先查看目前正在使用的php版本，【php  -v】

3.查看项目正在使用的php版本

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI3Njc3NjE_size_16_color_FFFFFF_t_70][]

### 二、[安装][Linux_Redis][thinkphp-queue][Linux_Redis] ###

【官方】 安装 thinkphp-queue

composer install topthink/think-queue

【我使用的命令】

composer require topthink/think-queue:2.\*

(首先确保已安装composer)

**注意：** \# Thinkphp5.1  
            composer require topthink/think-queue:2.\*  
            \# Thinkphp6  
            composer require topthink/think-queue:3.\*

### 三、[代码示例][Linux_Redis] ###

**[参考][Linux_Redis]  【thinkphp-queue 笔记】**： [https://github.com/coolseven/notes/tree/master/thinkphp-queue][https_github.com_coolseven_notes_tree_master_thinkphp-queue]

参考：[https://www.jianshu.com/p/6673f2d40581][https_www.jianshu.com_p_6673f2d40581]  
参考：[https://www.jianshu.com/p/588fb9c13493][https_www.jianshu.com_p_588fb9c13493]

先通过一段代码，了解一下 thinkphp-queue 的基本使用流程。

### 目标： ###

在业务控制器中推送一个新消息到一个名为 ‘helloJobQueue’ 的队列中，该消息中包含我们自定义的业务数据，然后，编写一个名为 Hello 的消费者类，并通过命令行去调用该消费者类获取这个消息，拿到定义的数据。

1.1 安装 thinkphp-queue

1.2 搭建消息队列的存储环境     【使用 Redis 】 安装并启动 Redis 服务

1.3 配置消息队列的驱动

安装完thinkphp-queue后会自动生成`\application\config\queue.php`

根据选择的存储方式，在 `\application\config\queue.php` 这个配置文件中，添加消息队列对应的驱动配置

return [
           'connector'  => 'Redis',		// Redis 驱动
           'expire'     => 60,		// 任务的过期时间，默认为60秒; 若要禁用，则设置为 null 
           'default'    => 'default',		// 默认的队列名称
           'host'       => '127.0.0.1',	// redis 主机ip
           'port'       => 6379,		// redis 端口
           'password'   => '',		// redis 密码
           'select'     => 0,		// 使用哪一个 db，默认为 db0
           'timeout'    => 0,		// redis连接的超时时间
           'persistent' => false,		// 是否是长连接
         
       //    'connector' => 'Database',   // 数据库驱动
       //    'expire'    => 60,           // 任务的过期时间，默认为60秒; 若要禁用，则设置为 null
       //    'default'   => 'default',    // 默认的队列名称
       //    'table'     => 'jobs',       // 存储消息的表名，不带前缀
       //    'dsn'       => [],
    
       //    'connector'   => 'Topthink',	// ThinkPHP内部的队列通知服务平台 ，本文不作介绍
       //    'token'       => '',
       //    'project_id'  => '',
       //    'protocol'    => 'https',
       //    'host'        => 'qns.topthink.com',
       //    'port'        => 443,
       //    'api_version' => 1,
       //    'max_retries' => 3,
       //    'default'     => 'default',
    
       //    'connector'   => 'Sync',		// Sync 驱动，该驱动的实际作用是取消消息队列，还原为同步执行
       ];

**    1.3.1 配置文件中的 expire 参数说明**

expire 参数指的是任务的过期时间, 单位为秒。 过期的任务，其准确的定义是

1.  任务的状态为执行中
2.  任务的开始执行的时刻 + expire > 当前时刻

expire 不为`null` 时 ，thinkphp-queue 会在每次获取下一个任务之前检查并重发过期(执行超时)的任务。

expire 为`null` 时，thinkphp-queue 不会检查过期的任务，性能相对较高一点。但是需要注意：

*   这些执行超时的任务会一直留在消息队列中，需要开发者另行处理(删除或者重发)！

**        对expire 参数理解或者使用不当时，很容易产生一些bug**，后面会举例提到。

1.4 消息的创建与推送

我们在业务控制器中创建一个新的消息，并推送到 `helloJobQueue` 队列

新增 `\application\index\controller\JobTest.php` 控制器，在该控制器中添加 `actionWithHelloJob` 方法

<?php
    /**
    * 文件路径： \application\index\controller\JobTest.php
    * 该控制器的业务代码中借助了thinkphp-queue 库，将一个消息推送到消息队列
    */
    namespace app\index\controller;
      use think\Exception;
    
      use think\Queue;
    
      class JobTest {
      /**
       * 一个使用了队列的 action
       */
      public function actionWithHelloJob(){
          
          // 1.当前任务将由哪个类来负责处理。 
          //   当轮到该任务时，系统将生成一个该类的实例，并调用其 fire 方法
          $jobHandlerClassName  = 'app\index\job\Hello'; 
          
          // 2.当前任务归属的队列名称，如果为新队列，会自动创建
          $jobQueueName  	  = "helloJobQueue"; 
          
          // 3.当前任务所需的业务数据 . 不能为 resource 类型，其他类型最终将转化为json形式的字符串
          //   ( jobData 为对象时，存储其public属性的键值对 )
          $jobData       	  = [ 'ts' => time(), 'bizId' => uniqid() , 'a' => 1 ] ;
          
          // 4.将该任务推送到消息队列，等待对应的消费者去执行
          $isPushed = Queue::push( $jobHandlerClassName , $jobData , $jobQueueName );	
          
          // database 驱动时，返回值为 1|false  ;   redis 驱动时，返回值为 随机字符串|false
          if( $isPushed !== false ){  
              echo date('Y-m-d H:i:s') . " a new Hello Job is Pushed to the MQ"."<br>";
          }else{
              echo 'Oops, something went wrong.';
          }
      }
     }

**注意:** 在这个例子当中，我们是手动指定的 `$jobHandlerClassName` ，更合理的做法是先定义好消息名称与消费者类名的映射关系，然后由某个可以获取该映射关系的类来推送这个消息。这样，生产者只需要知道消息的名称，而无需指定哪个消费者类来处理。

> 除了 `Queue::push( $jobHandlerClassName , $jobData , $jobQueueName );`这种方式之外，还可以直接传入 `Queue::push( $jobHandlerObject ,null , $jobQueueName );` 这时，需要在 $jobHandlerObject 中定义一个 `handle()` 方法，消息队列在执行到该任务时会自动反序列化该对象，并调用其 `handle()`方法。 该方式中, 数据需要提前挂载在 `$jobHandlerObject` 对象上。

1.5 消息的消费与删除

编写 Hello 消费者类，用于处理 `helloJobQueue` 队列中的任务

新增 `\application\index\job\Hello.php` 消费者类，并编写其 `fire()` 方法

<?php
      /**
       * 文件路径： \application\index\job\Hello.php
       * 这是一个消费者类，用于处理 helloJobQueue 队列中的任务
       */
      namespace app\index\job;
    
      use think\queue\Job;
    
      class Hello {
          
          /**
           * fire方法是消息队列默认调用的方法
           * @param Job            $job      当前的任务对象
           * @param array|mixed    $data     发布任务时自定义的数据
           */
          public function fire(Job $job,$data)
          {
              // 有些消息在到达消费者时,可能已经不再需要执行了
              $isJobStillNeedToBeDone = $this->checkDatabaseToSeeIfJobNeedToBeDone($data);
              if(!$isJobStillNeedToBeDone){
                  $job->delete();
                  return;
              }
            
              $isJobDone = $this->doHelloJob($data);
            
              if ($isJobDone) {
                  // 如果任务执行成功， 记得删除任务
                  $job->delete();
                  print("<info>Hello Job has been done and deleted"."</info>\n");
              }else{
                  if ($job->attempts() > 3) {
                      //通过这个方法可以检查这个任务已经重试了几次了
                      print("<warn>Hello Job has been retried more than 3 times!"."</warn>\n");
                      
      				        $job->delete();
                      
                      // 也可以重新发布这个任务
                      //print("<info>Hello Job will be availabe again after 2s."."</info>\n");
                      //$job->release(2); //$delay为延迟时间，表示该任务延迟2秒后再执行
                  }
              }
          }
          
          /**
           * 有些消息在到达消费者时,可能已经不再需要执行了
           * @param array|mixed    $data     发布任务时自定义的数据
           * @return boolean                 任务执行的结果
           */
          private function checkDatabaseToSeeIfJobNeedToBeDone($data){
              return true;
          }
    
          /**
           * 根据消息中的数据进行实际的业务处理...
           */
          private function doHelloJob($data) 
          {
              print("<info>Hello Job Started. job Data is: ".var_export($data,true)."</info> \n");
              print("<info>Hello Job is Fired at " . date('Y-m-d H:i:s') ."</info> \n");
              print("<info>Hello Job is Done!"."</info> \n");
              
              return true;
          }
      }

至此，所有的代码都已准备完毕，在运行消息队列之前，我们先看一下现在的目录结构：

[![目录结构-代码示例][-]][-_-]

1.6 发布任务

在浏览器中访问 [http://your.project.domain/index/job\_test/actionWithHelloJob][http_your.project.domain_index_job_test_actionWithHelloJob] ,可以看到消息推送成功。

[![浏览器提示消息推送结果][3dce370ba0d48e7fff13f9e0c8c8ff1b.png]][3dce370ba0d48e7fff13f9e0c8c8ff1b.png 1]

1.7 处理任务

切换当前终端窗口的目录到项目根目录下，执行

php think queue:work --queue helloJobQueue

可以看到执行的结果类似如下:

[![命令行执行结果][294114db25ef0e96cce0a6d852c47231.png]][294114db25ef0e96cce0a6d852c47231.png 1]

至此，我们成功地经历了一个消息的 创建 -> 推送 -> 消费 -> 删除 的基本流程

下文，将介绍 thinkphp-queue 的详细使用方法。如配置介绍，基本原理，各种特殊情况的处理等

### 二 详细介绍 ###

2.1 命令模式

*  **queue:subscribe 命令** \[官方未提供示例, 略过\]
 *  **queue:work 命令**
    
    work 命令： 该命令将启动一个 work 进程来处理消息队列。
    
        php think queue:work --queue helloJobQueue
 *  **queue:listen 命令**
    
    listen 命令： 该命令将会启动一个 listen 进程 ，然后由 listen 进程通过 `proc_open(‘php think queue:work --queue="%s" --delay=%s --memory=%s --sleep=%s --tries=%s’)` 的方式来周期性地创建一次性的 work 进程来消费消息队列, 并且限制该 work 进程的执行时间, 同时通过管道来监听 work 进程的输出，。
    
        php think queue:listen --queue helloJobQueue

2.2 命令行参数

*  Work 模式
    
        php think queue:work \
        --daemon            //是否循环执行，如果不加该参数，则该命令处理完下一个消息就退出
        --queue  helloJobQueue  //要处理的队列的名称
        --delay  0 \        //如果本次任务执行抛出异常且任务未被删除时，设置其下次执行前延迟多少秒,默认为0
        --force  \          //系统处于维护状态时是否仍然处理任务，并未找到相关说明
        --memory 128 \      //该进程允许使用的内存上限，以 M 为单位
        --sleep  3 \        //如果队列中无任务，则sleep多少秒后重新检查(work+daemon模式)或者退出(listen或非daemon模式)
        --tries  2          //如果任务已经超过尝试次数上限，则触发‘任务尝试次数超限’事件，默认为0
 *  Listen 模式
    
        php think queue:listen \
        --queue  helloJobQueue \   //监听的队列的名称
        --delay  0 \         //如果本次任务执行抛出异常且任务未被删除时，设置其下次执行前延迟多少秒,默认为0
        --memory 128 \       //该进程允许使用的内存上限，以 M 为单位
        --sleep  3 \         //如果队列中无任务，则多长时间后重新检查
        --tries  0 \         //如果任务已经超过重发次数上限，则进入失败处理逻辑，默认为0
        --timeout 60         // work 进程允许执行的最长时间，以秒为单位
    
    可以看到 listen 模式下，不包含 `--deamon` 参数，原因下面会说明

2.3 work 模式和 listen 模式的区别

两者都可以用于处理消息队列中的任务

区别在于:

*  **2.3.1 执行原理不同**
    
     *  work 命令是**单进程**的处理模式。
        
        按照是否设置了 `--daemon` 参数，work命令又可分为单次执行和循环执行两种模式。
        
         *  单次执行：不添加 `--daemon`参数，该模式下,work进程在处理完下一个消息后直接结束当前进程。当队列为空时，会sleep一段时间然后退出。
         *  循环执行：添加了 `--daemon`参数，该模式下,work进程会循环地处理队列中的消息，直到内存超出参数配置才结束进程。当队列为空时，会在每次循环中sleep一段时间。
     *  listen 命令是 **双进程 + 管道** 的处理模式。
        
        listen命令所在的进程会循环地创建 **单次执行模式的 work 进程**，每次创建的 work 进程只消费一个消息就会结束, 然后 listen 进程再创建一个新的 work 进程，
        
         *  listen 进程会定时检查当前的 work 进程执行时间是否超过了 --timeout 参数的值, 如果已超时, 则 listen 进程会 kill 掉 work 进程, 然后抛出异常
         *  listen 进程会通过管道来监听当前的 work 进程的输出, 当 work 进程有输出时, listen 进程会将输出写入到 stdout / stderr
         *  listen 进程会定时通过 `proc_get_status()` 来监控当前的 work 进程是否仍在运行, work 进程消费完一个任务之后, work 进程就结束了，其状态会变成 terminated, 此时 listen 进程就会重新创建一个新的 work 进程并对其计时, 新的 work 进程开始消费下一个任务
 *  **2.3.2 结束时机不同**
    
     *  work 命令的结束时机在上面的执行原理部分已叙述，此处不再重复
     *  listen 命令中，listen 进程和 work 进程会在以下情况下结束：
        
         *  listen 进程会定时检查当前的 work 进程的执行时间是否超过了 --timeout 参数的值，如果已超时, 此时 listen 进程会先 kill 掉当前的 work 进程, 然后抛出一个 `ProcessTimeoutException` 异常并结束 listen 进程
         *  listen 进程会定时检查自身使用的内存是否超过了 `--memory` 参数的值，如果已超过, 此时 listen 进程会直接 die 掉, work 进程也会自动结束.
 *  **2.3.3 性能不同**
    
     *  work 命令是在脚本内部做循环，框架脚本在命令执行的初期就已加载完毕；
     *  而listen模式则是处理完一个任务之后新开一个work进程，此时会重新加载框架脚本。
        
        因此： **work 模式的性能会比listen模式高**。
        
        注意：当代码有更新时，work 模式下需要手动去执行 `php think queue:restart` 命令重启队列来使改动生效；而listen 模式会自动生效,无需其他操作。
 *  **2.3.4 超时控制能力**
    
     *  work 模式本质上既不能控制进程自身的运行时间，也无法限制执行中的任务的执行时间。
        
        举例来说，假如你在某次上线之后，在上文中的 `\application\index\job\Hello.php` 消费者的`fire`方法中添加了一段死循环 ：
        
            public function fire(){
               while(true){ //死循环
                   $consoleOutPut->writeln("<info>I am looping forever inside a job.</info> \n");
                   sleep(1);
               }
            }
        
        那么这个循环将永远不能停止，直到任务所在的进程超过内存限制或者由管理员手动结束。这个过程不会有任何的告警。更严重的是，如果你配置了expire ，那么这个死循环的任务可能会污染到同样处理 `helloJobQueue` 队列的其他work进程，最后好几个work进程将被卡死在这段死循环中。详情后文会说明。
        
        **work 模式下的超时控制能力，实际上应该理解为 多个work 进程配合下的过期任务重发能力。**
     *  而 listen 命令可以限制 listen 进程创建的 work 进程的最大执行时间。
        
        listen 命令可通过 `--timeout` 参数限制 work 进程允许运行的最长时间，超过该时间限制后, work 进程会被强制 kill 掉, listen 进程本身也会抛出异常并结束；
     *  这里有必要补充一下 expire 和 timeout 之间的区别：
        
         *  expire 在配置文件中设置，timeout 在 listen命令 的命令行参数中设置，而且，expire 和 timeout 是两个不同层次上的概念：
         *  expire 是指任务的过期时间。这个时间是全局的，影响到所有的work进程。(不管是独立的work命令还是 listen 模式下创建的的 work 进程) 。expire 针对的对象是 **任务**。
         *  timeout 是指 work 进程的超时时间。这个时间只对当前执行的 listen 命令有效。timeout 针对的对象是 **work 进程**。
 *  **2.3.5 使用场景不同**
    
    根据上面的介绍，可以看到，
    
    work 命令的适用场景是：
    
     *  任务数量较多
     *  性能要求较高
     *  任务的执行时间较短
     *  消费者类中不存在死循环，sleep() ，exit() ,die() 等容易导致bug的逻辑
    
    listen命令的适用场景是：
    
     *  任务数量较少
     *  任务的执行时间较长(如生成大型的excel报表等)，
     *  任务的执行时间需要有严格限制

2.4 消息队列的开始，停止与重启

*  开始一个消息队列：
    
        php think queue:work
 *  停止所有的消息队列：
    
        php think queue:restart
 *  重启所有的消息队列：
    
        php think queue:restart 
        php think queue:work

2.5 多模块，多任务的处理

*  多模块
    
    > 单模块项目推荐使用 `app\job` 作为任务类的命名空间
    > 
    > 多模块项目可用使用 `app\module\job` 作为任务类的命名空间 也可以放在任意可以自动加载到的地方
 *  多任务
    
    > 如果一个任务类里有多个小任务的话，在发布任务时，需要用 `任务的类名@方法名` 如 `app\lib\job\Job2@task1`、`app\lib\job\Job2@task2`
    > 
    > 注意：命令行中的 --queue 参数不支持@解析
    
    多任务例子:
    
     *  在 `\application\index\controller\JobTest.php` 控制器中，添加 `actionWithMultiTask()`方法：
    
        public function actionWithMultiTask(){
            	
          $taskType = $_GET['taskType'];
            switch ($whichTask) {
               case 'taskA':
                   $jobHandlerClassName  = 'application\index\job\MultiTask@taskA';
                   $jobDataArr = ['a'	=> '1'];
                   $jobQueueName = "multiTaskJobQueue";	
                   break;
               case 'taskB':
                   $jobHandlerClassName  = 'application\index\job\MultiTask@taskB';
                   $jobDataArr = ['b'	=> '2'];
                   $jobQueueName = "multiTaskJobQueue";		
                   break;
                default:
                   break;
           }
          
          $isPushed = Queue::push($jobHandlerClassName, $jobDataArr, $jobQueueName);
          if ($isPushed !== false) {
            echo("the $taskType of MultiTask Job has been Pushed to ".$jobQueueName ."<br>");
          }else{
            throw new Exception("push a new $taskType of MultiTask Job Failed!");
          }
        }
    
     *  新增 `\application\index\job\MultiTask.php` 消费者类，并编写其 `taskA()` 和 `taskB()`方法
    
        <?php
        /**
         * 文件路径： \application\index\job\MultiTask.php
         * 这是一个消费者类，用于处理 multiTaskJobQueue 队列中的任务
         */
        namespace app\index\job;
        
        use think\queue\Job;
        
        class MultiTask {
         
            public function taskA(Job $job,$data){
                
                $isJobDone = $this->_doTaskA($data);
        
                if ($isJobDone) {
                    $job->delete();
                    print("Info: TaskA of Job MultiTask has been done and deleted"."\n");
                }else{
                    if ($job->attempts() > 3) {
                        $job->delete(); 	
                    }
                }
            }
          
            public function taskB(Job $job,$data){
                
                $isJobDone = $this->_doTaskA($data);
        
                if ($isJobDone) {
                    $job->delete();
                    print("Info: TaskB of Job MultiTask has been done and deleted"."\n");
                }else{
                    if ($job->attempts() > 2) {
                        $job->release(); 	
                    }
                }
            }
        
            private function _doTaskA($data) {
                print("Info: doing TaskA of Job MultiTask "."\n");
                return true;
            }
          
            private function _doTaskB($data) {
                print("Info: doing TaskB of Job MultiTask "."\n");
                return true;
            }

2.6 消息的延迟执行与定时执行

延迟执行，相对于即时执行，是用来限制某个任务的最早可执行时刻。在到达该时刻之前，该任务会被跳过。

可以利用该功能实现**定时任务**。

使用方式：

*  在生产者业务代码中：

// 即时执行
    $isPushed = Queue::push($jobHandlerClassName, $jobDataArr, $jobQueueName);
    
    // 延迟 2 秒执行
    $isPushed = Queue::later( 2, $jobHandlerClassName, $jobDataArr, $jobQueueName);
    
    // 延迟到 2017-02-18 01:01:01 时刻执行
    $time2wait = strtotime('2017-02-18 01:01:01') - strtotime('now');	
    $isPushed = Queue::later($time2wait,$jobHandlerClassName, $jobDataArr, $jobQueueName);

*  在消费者类中：

// 重发，即时执行
    $job->release();
    
    // 重发，延迟 2 秒执行
    $job->release(2);
    
    // 延迟到 2017-02-18 01:01:01 时刻执行
    $time2wait = strtotime('2017-02-18 01:01:01') - strtotime('now');
    $job->release($time2wait);

*  在命令行中：

//如果消费者类的fire()方法抛出了异常且任务未被删除时，将自动重发该任务，重发时，会设置其下次执行前延迟多少秒,默认为0
    php think queue:work --delay 3

2.7 消息的重发

thinkphp-queue 中，消息的重发时机有3种：

*  2.7.1 在消费者类中手动重发：

if( $isJobDone === false){
        $job->release();
    }

*  2.7.2 work进程自动重发，需同时满足以下两个条件
    
     *  消费者类的 fire() 方法抛出了异常
     *  任务未被删除
 *  2.7.3 当配置了 expire 不为 `null` 时，work 进程内部每次查询可用任务之前，会先自动重发已过期的任务。

> 补充：
> 
> 在database 模式下，2.7.1 和 2.7.2 中的重发逻辑是先删除原来的任务，然后插入一个新的任务。2.7.3 中的重发时机是直接更新原任务。
> 
> 而在redis 模式下，3种重发都是先删除再插入。
> 
> 不管是哪种重发方式，重发之后，任务的已尝试次数会在原来的基础上 +1 。

此外，消费者类中需要注意，如果 `fire()` 方法中可能抛出异常，那么

*  如果不需要自动重发的话， 请在抛出异常之前将任务删除 `$job->delete()` ，否则会被框架自动重发。
 *  如果需要自动重发的话，请直接抛出异常，不要在 `fire()` 方法中又手动使用 `$job->release()` , 这样会导致该任务被重发两次，产生两个一样的新任务。

2.8 任务的失败回调及告警

当同时满足以下条件时，将触发任务失败回调：

*  命令行的 `--tries` 参数的值大于0
 *  任务的已尝试次数大于 命令行的 `--tries` 参数
 *  开发者添加了 `queue_failed` 事件标签及其对应的回调代码
 *  消费者类中定义了 `failed()` 方法，用于接收任务失败的通知

注意， `queue_failed` 标签需要在安装了 `thinkphp-queue `之后 **手动** 去 `\application\tags.php` 文件中添加。

首先，我们添加 `queue_failed` 事件标签, 及其对应的回调方法

// 文件路径： \application\tags.php
    // 应用行为扩展定义文件
    return [
        // 应用初始化
        'app_init'     => [],
        // 应用开始
        'app_begin'    => [],
        // 模块初始化
        'module_init'  => [],
        // 操作开始执行
        'action_begin' => [],
        // 视图内容过滤
        'view_filter'  => [],
        // 日志写入
        'log_write'    => [],
        // 应用结束
        'app_end'      => [],
    
        // 任务失败统一回调,有四种定义方式
        'queue_failed'=> [
          
             // 数组形式，[ 'ClassName' , 'methodName']
            ['application\\behavior\\MyQueueFailedLogger', 'logAllFailedQueues']
             
             // 字符串(静态方法)，'StaicClassName::methodName'
             // 'MyQueueFailedLogger::logAllFailedQueues'   
          
             // 字符串(对象方法)，'ClassName'，此时需在对应的ClassName类中添加一个名为 queueFailed 的方法
             // 'application\\behavior\\MyQueueFailedLogger'
             
             // 闭包形式
             /*
             function( &$jobObject , $extra){
                 // var_dump($jobObject);
                 return true;
             }
             */
        ]
    ];

这里，我们选择数组形式的回调方式，新增 `\application\behavior\MyQueueFailedLogger` 类，添加一个 `logAllFailedQueues()` 方法

<?php
    /**
     * 文件路径: \application\behavior\MyQueueFailedLogger.php
     * 这是一个行为类，用于处理所有的消息队列中的任务失败回调
     */
    
    namespace app\behavior;
    
    
    class MyQueueFailedLogger {
           
        const should_run_hook_callback = true;
      
        /**
         * @param $jobObject   \think\queue\Job   //任务对象，保存了该任务的执行情况和业务数据
         * @return bool     true                  //是否需要删除任务并触发其failed() 方法
         */
        public function logAllFailedQueues(&$jobObject){
            
            $failedJobLog = [
                'jobHandlerClassName'   => $jobObject->getName(), // 'application\index\job\Hello'
                'queueName' => $jobObject->getQueue(),			   // 'helloJobQueue'	 
                'jobData'   => $jobObject->getRawBody()['data'],  // '{'a': 1 }'
                'attempts'  => $jobObject->attempts(),            // 3
            ];
            var_export(json_encode($failedJobLog,true));
            
               // $jobObject->release();     //重发任务
              //$jobObject->delete();         //删除任务
              //$jobObject->failed();	  //通知消费者类任务执行失败
            
            return self::should_run_hook_callback;         
        }
    }

需要注意该回调方法的返回值：

*  返回 true 时，系统会自动删除该任务，并且自动调用消费者类中的 `failed()` 方法
 *  返回 false 时，系统不会自动删除该任务，也不会自动调用消费者类中的 `failed()` 方法，需要开发者另行处理失败任务的删除和通知。

最后，在消费者类中，添加 `failed()` 方法

/**
     * 文件路径： \application\index\job\HelloJob.php
     */
    
    /**
     * 该方法用于接收任务执行失败的通知，你可以发送邮件给相应的负责人员
     * @param $jobData  string|array|...      //发布任务时传递的 jobData 数据
     */
    public function failed($jobData){
        send_mail_to_somebody() ; 
        
        print("Warning: Job failed after max retries. job data is :".var_export($data,true)."\n"; 
    }

这样，就可以做到任务失败的**记录**与**告警**

2.9 处理过期的任务

过期这个概念用文字比较难描述清楚，建议先看一下 **深入理解** 中 **3.4 消息处理的详细流程图**

### 三 深入理解 ###

3.1 thinkphp-queue 中消息与队列的保存方式

*  Redis
    
    在 Redis 中，每一个 队列 都三个key 与之对应 ，以 helloJobQueue 队列举例，其在redis 中的保存方式为：
    
    <table> 
     <thead> 
      <tr> 
       <th>key名</th> 
       <th>类型</th> 
       <th>说明</th> 
      </tr> 
     </thead> 
     <tbody> 
      <tr> 
       <td>queues:helloJobQueue</td> 
       <td>List ， 列表</td> 
       <td>待执行的任务列表</td> 
      </tr> 
      <tr> 
       <td>queues:helloJobQueue:delayed</td> 
       <td>Sorted Set，有序集合</td> 
       <td>延迟执行和定时执行的任务集合</td> 
      </tr> 
      <tr> 
       <td>queues:helloJobQueue:reserved</td> 
       <td>Sorted Set，有序集合</td> 
       <td>执行中的任务集合</td> 
      </tr> 
     </tbody> 
    </table>
    
    > 使用的`:`分隔符, 只是用来表示相关key的关联性。本身没有特殊含义。使用分隔符是一种常见的组织key的方式。
    
    其中，在`queues:helloJobQueue` 列表中，每个元素的形式如下：
    
    [![redis中的队列-queue][redis_-queue]][redis_-queue_redis_-queue]
    
    在 `queues:helloJobQueue:delayed` 和 `queues:helloJobQueue:delayed` 有序集合中，每个元素的形式如下：
    
    [![redis中的队列-queue-reserved][redis_-queue-reserved]][redis_-queue-reserved_redis_-queue-reserved]
    
    可以看到，在有序集合中，每个元素代表一个任务，该元素的 Score 为该任务的入队时间戳，任务的 value 为json 格式，保存了任务的执行情况和业务数据。将value decode 为数组后形式如下：
    
        [
          'job'  => 'application\\index\\job\\Hello' ,  // jobHandlerClassName，消费者类的类名 
          'data' => [					  // 生产者传入的业务数据
             'time' => '2017-02-18 16:20:10',
             'data' => 'I have 648 apples'
          ],
          'id'   => '77IasdasadIasdadadadKL8t',	// 一个随机的32位字符串
          'attempts' => 2				// 任务的已尝试次数
        ]
    
    redis驱动下，为了实现任务的延迟执行和过期重发，任务将在这三个key中来回转移，详情可见 3.5
 *  Database
    
    在 Database 中，每个任务对应到表中的一行，queue 字段用来区分不同的队列。
    
    表的字段结构如下:
    
    [![数据库字段说明][74935cfc2748154c526d2cf7cfcd571b.png]][74935cfc2748154c526d2cf7cfcd571b.png 1]
    
    其中，payLoad 字段保存了消息的执行者和业务数据，payLoad 字段采用 json 格式的字符串来保存消息，将其 decode 为数组后形式如下：
    
        [
         'job'   => 'application\\index\\job\\Hello', // jobHandlerClassName，消费者类的类名 
         'data'  => string|array|integer|object       // 生产者传入的业务数据
        ]

3.2 thinkphp-queue 的目录结构和类关系图

[![thinkphp-queue的文件目录][thinkphp-queue 1]][thinkphp-queue_thinkphp-queue 1]

这些类构成了消息队列中的几个角色：

<table> 
 <thead> 
  <tr> 
   <th>角色</th> 
   <th>类名</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>命令行</td> 
   <td>Command + Worker</td> 
   <td>负责解析命令行参数，控制队列的启动，重启</td> 
  </tr> 
  <tr> 
   <td>驱动</td> 
   <td>Queue + Connector</td> 
   <td>负责队列的创建，以及消息的入队，出队等操作</td> 
  </tr> 
  <tr> 
   <td>任务</td> 
   <td>Job</td> 
   <td>用于将消息转化为一个任务对象，供消费者使用</td> 
  </tr> 
  <tr> 
   <td>生产者</td> 
   <td>业务代码</td> 
   <td>负责消息的创建与发布</td> 
  </tr> 
  <tr> 
   <td>消费者</td> 
   <td>业务代码</td> 
   <td>负责任务的接收与执行</td> 
  </tr> 
 </tbody> 
</table>

各个类之间的关系图如下：

thinkphp-queue类关系图

3.3 Deamon模式的执行流程

Daemon模式与非daemon模式状态图

3.4 Database模式下消息处理的详细流程

下图中，展示了database 模式下消息处理的详细流程，redis 驱动下大体类似

Database模式下消息获取和执行的具体流程

3.5 redis 驱动下的任务重发细节

在redis驱动下，为了实现任务的延迟执行和过期重发，任务将在这三个key中来回转移。

在3.4 Database模式下消息处理的消息流程中，我们知道，如果配置的expire 不是null ，那么 thinkphp-queue的work进程每次在获取下一个可执行任务之前，会先尝试重发所有过期的任务。而在redis驱动下，这个步骤则做了更多的事情，详情如下：

1.  从 `queue:xxx:delayed` 的key中查询出有哪些任务在当前时刻已经可以开始执行，然后将这些任务转移到 `queue:xxx` 的key的尾部。
2.  从 `queue:xxx:reserved` 的key中查询出有哪些任务在当前时刻已经过期，然后将这些任务转移到 `queue:xxx`的key的尾部。
3.  尝试从 `queue:xxx` 的key的头部取出一个任务，如果取出成功，那么，将这个任务转移到 `queue:xxx:reserved` 的key 的头部，同时将这个任务实例化成任务对象，交给消费者去执行。

用图来表示这个步骤的具体过程如下：

redis队列中的过期任务重发步骤--执行前：

[![redis队列中的过期任务重发步骤-执行前][redis_-]][redis_-_redis_-]

redis队列中的过期任务重发步骤--执行后：

[![redis队列中的过期任务重发步骤--执行后][redis_--]][redis_--_redis_--]

3.6 thinkphp-queue的性能

*  测试环境 :
    
    虚拟机 Ubuntu 16.04 ， PHP 7.1 ，TP5，Redis 3.2 ， 双核 I5 6400，3G 内存
 *  测试方式 :
    
    使用 Redis 驱动，在一个控制器中循环推送 40000 条消息到消息队列；
    
    使用`php think queue:work --daemon`去消费这些消息，计算推送和消费各自所耗的时间。
 *  测试结果 :
    
    在最简单的逻辑下，平均每秒中可推送8000个消息，平均每秒可消费200个消息。

**注意：由于在测试时，Host 机本身的cpu和内存长期100%，并且虚拟机中的各项服务并未专门调优，因此该测试结果并不具备参考性**。

3.7 thinkphp-queue 的注意事项

*  **3.7.1** 使用建议
    
     *  **任务完成后, 使用 `$job->delete()` 删除任务**
     *  在消费者类的 `fire()` 方法中，使用 `$job->attempt()` 检查任务已执行次数，对于次数异常的，作相应的处理。
     *  在消费者类的 `fire()` 方法中，根据业务数据来判断该任务是否已经执行过，以避免该任务被重复执行。
     *  编写失败回调事件，将事件中失败的任务及时通知给开发人员。
 *  **3.7.2** 使用了 `queue:work --daemon` ，但更新代码后没有使用 `queue:restart` 重启 work 进程, 使得 work 进程中的代码与最新的代码不同，出现各种问题。
 *  **3.7.3** 使用了 `queue:work --daemon` ，但是消费者类的 fire() 方法中存在死循环，或 `sleep(n)` 等逻辑，导致消息队列被堵塞；或者使用了 `exit()` , `die()` 这样的逻辑，导致work进程直接终止 。
 *  **3.7.4** 配置的 expire 为`null` ，但并没有自行处理过期的任务，导致过期的任务得不到处理，且一直占用消息队列的空间。
 *  **3.7.5** 配置的 expire `不为null` ，但配置的 expire 时间太短，以至于 expire 时间 < 消费者的 `fire()` 方法所需时间 + 删除该任务所需的时间 ，那么任务将被误认为执行超时，从而被消息队列还原为待执行状态。
 *  **3.7.6** queue:work 属于 cli 模式, 其入口文件是根目录下的 think 文件, 网页访问属于 sapi 模式, 其入口是 public 目录下的 index.php 文件. 如果需要在入口文件中定义自定义常量, 记得去修改对应的入口文件. 具体可参考 chichoyi 的说明 [https://github.com/coolseven/notes/issues/7][https_github.com_coolseven_notes_issues_7]

# 四 拓展 #

4.1 队列的稳定性和拓展性

*  稳定性：不管是 listen 模式还是 work 模式，都建议使用 supervisor 或者 自定义的cron 脚本，去定时检查 work 进程是否正常
 *  拓展性： 当某个队列的消费者不足时，再给这个队列添加 work进程即可。
 *  队列管理，队列的列表，队列的 work 进程数量控制，队列的任务数量变化趋势 //TBD
 *  任务管理，任务的列表，添加/**撤回**/查询任务，修改任务的 执行者/执行时间/优先级/数据 等 //TBD

4.2 编写自定义的 thinkphp-queue 驱动

//TBD

4.3 编写消息队列的单元测试

//TBD

4.4 与其他PHP消息队列库的对比

TP5的消息队列与Laravel的消息队列比较相似，下面是与laravel 中的消息队列的一些对比：

<table> 
 <thead> 
  <tr> 
   <th>&nbsp;</th> 
   <th>thinkphp-queue (v1.1.2)</th> 
   <th>laravel-queue (v5.3)</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>内置的驱动</td> 
   <td>Database，Redis，Sync，TopThink</td> 
   <td>Database，Redis, Sync(在laravel中称为 null)。</td> 
  </tr> 
  <tr> 
   <td>Redis驱动要求</td> 
   <td>安装redis的C扩展</td> 
   <td>安装 predis 包 + LUA脚本</td> 
  </tr> 
  <tr> 
   <td>推送任务</td> 
   <td>允许推送 消费者类名，消费者对象</td> 
   <td>允许推送消费者类名，消费者对象，闭包</td> 
  </tr> 
  <tr> 
   <td>失败任务处理</td> 
   <td>触发失败回调事件 (有Bug)</td> 
   <td>触发失败回调事件 + 移动任务到 failed_jobs表?</td> 
  </tr> 
  <tr> 
   <td>消息订阅</td> 
   <td>subscribe 命令+ Topthink驱动(注：未实现/未提供)</td> 
   <td>subscribe 命令 + 安装IronMQ 驱动</td> 
  </tr> 
  <tr> 
   <td>删除任务</td> 
   <td>消费者类中手动删除</td> 
   <td>任务完成后自动删除</td> 
  </tr> 
  <tr> 
   <td>推送到多个队列</td> 
   <td>需自己实现</td> 
   <td>原生支持</td> 
  </tr> 
  <tr> 
   <td>延迟执行</td> 
   <td>支持 (有Bug)</td> 
   <td>支持</td> 
  </tr> 
  <tr> 
   <td>消息重发</td> 
   <td>支持</td> 
   <td>支持</td> 
  </tr> 
  <tr> 
   <td>检查已执行次数</td> 
   <td>原生支持</td> 
   <td>需在消费者类中显式 use 相关的 trait</td> 
  </tr> 
  <tr> 
   <td>执行方式</td> 
   <td>work 模式 + listen 模式</td> 
   <td>work 模式 + listen 模式</td> 
  </tr> 
  <tr> 
   <td>进程命令</td> 
   <td>开启，停止，重启</td> 
   <td>开启，停止，重启</td> 
  </tr> 
  <tr> 
   <td>任务命令</td> 
   <td>无</td> 
   <td>展示失败任务列表，重试某个失败任务，删除某个失败任务</td> 
  </tr> 
  <tr> 
   <td>支持的事件</td> 
   <td>失败回调事件</td> 
   <td>失败回调事件，支持消费前事件，消费后事件</td> 
  </tr> 
 </tbody> 
</table>

[thinkphp-queue]: https://github.com/top-think/think-queue/releases
[Linux_Redis]: https://www.cnblogs.com/ytkah/p/9092378.html
[1f3c50a1d4c0ee865311bd239625dfa7.png]: /images/20221123/cf20e7d0185b497d86519e61ec86c5f5.png
[0fc71b3f36025c52bd144fb22b300eb6.png]: /images/20221123/9c913bd972d34b5b9414d7ca1ee61a11.png
[18dba49f11b5a68880c79ac0d4fb575f.png]: /images/20221123/b397609104cb4fff9214171eeeabf99c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI3Njc3NjE_size_16_color_FFFFFF_t_70]: /images/20221123/32a5bf566aa240e89edf056ce7c76dc6.png
[https_github.com_coolseven_notes_tree_master_thinkphp-queue]: https://github.com/coolseven/notes/tree/master/thinkphp-queue
[https_www.jianshu.com_p_6673f2d40581]: https://www.jianshu.com/p/6673f2d40581
[https_www.jianshu.com_p_588fb9c13493]: https://www.jianshu.com/p/588fb9c13493
[-]: https://img-blog.csdnimg.cn/img_convert/f92777a2331279466bc9ca675250d210.png
[-_-]: https://github.com/coolseven/notes/blob/master/thinkphp-queue/%E7%9B%AE%E5%BD%95%E7%BB%93%E6%9E%84-%E4%BB%A3%E7%A0%81%E7%A4%BA%E4%BE%8B.png
[http_your.project.domain_index_job_test_actionWithHelloJob]: http://your.project.domain/index/job_test/actionWithHelloJob
[3dce370ba0d48e7fff13f9e0c8c8ff1b.png]: https://img-blog.csdnimg.cn/img_convert/3dce370ba0d48e7fff13f9e0c8c8ff1b.png
[3dce370ba0d48e7fff13f9e0c8c8ff1b.png 1]: https://github.com/coolseven/notes/blob/master/thinkphp-queue/%E6%B5%8F%E8%A7%88%E5%99%A8%E6%8F%90%E7%A4%BA%E6%B6%88%E6%81%AF%E6%8E%A8%E9%80%81%E7%BB%93%E6%9E%9C.png
[294114db25ef0e96cce0a6d852c47231.png]: https://img-blog.csdnimg.cn/img_convert/294114db25ef0e96cce0a6d852c47231.png
[294114db25ef0e96cce0a6d852c47231.png 1]: https://github.com/coolseven/notes/blob/master/thinkphp-queue/%E5%91%BD%E4%BB%A4%E8%A1%8C%E6%89%A7%E8%A1%8C%E7%BB%93%E6%9E%9C.png
[redis_-queue]: https://img-blog.csdnimg.cn/img_convert/2a3bfeca090156b3a8a6a82bacee279b.png
[redis_-queue_redis_-queue]: https://github.com/coolseven/notes/blob/master/thinkphp-queue/redis%E4%B8%AD%E7%9A%84%E9%98%9F%E5%88%97-queue.png
[redis_-queue-reserved]: https://img-blog.csdnimg.cn/img_convert/242dac9b88adab67a7bd56b98f77d68a.png
[redis_-queue-reserved_redis_-queue-reserved]: https://github.com/coolseven/notes/blob/master/thinkphp-queue/redis%E4%B8%AD%E7%9A%84%E9%98%9F%E5%88%97-queue-reserved.png
[74935cfc2748154c526d2cf7cfcd571b.png]: https://img-blog.csdnimg.cn/img_convert/74935cfc2748154c526d2cf7cfcd571b.png
[74935cfc2748154c526d2cf7cfcd571b.png 1]: /images/20221123/1d6357c99c0847aebdc0964aec16311f.png
[thinkphp-queue 1]: https://img-blog.csdnimg.cn/img_convert/fc83d811ba44026b92fddf24d4d265f8.png
[thinkphp-queue_thinkphp-queue 1]: /images/20221123/d12b366e57a84f1db77d54d7d30cde55.png
[redis_-]: https://img-blog.csdnimg.cn/img_convert/b5bfb89fc4958b9ee361dd4095a7152f.png
[redis_-_redis_-]: /images/20221123/5760afb2dd9b4daebaa7f81ab22ac314.png
[redis_--]: https://img-blog.csdnimg.cn/img_convert/aa8c318948e753d358ba3a638bff2283.png
[redis_--_redis_--]: /images/20221123/e725ad33963844bcb6d55dee91d2e3b4.png
[https_github.com_coolseven_notes_issues_7]: https://github.com/coolseven/notes/issues/7