ThinkPHP基于think-queue的队列插件实现消息推送

think-queue是ThinkPHP官方提供的一个消息队列服务，是专门支持队列服务的扩展包。think-queue消息队列适用于大并发或返回结果时间比较长且需要批量操作的第三方接口，可用于短信发送、邮件发送、APP推送。think-queue消息队列可进行发布、获取、执行、删除、重发、失败处理、延迟执行、超时控制等操作。

前言

传统的程序执行流程一般是 即时|同步|串行的，在某些场景下，会存在并发低，吞吐量低，响应时间长等问题。在大型系统中，一般会引入消息队列的组件，将流程中部分任务抽离出来放入消息队列，并由专门的消费者作针对性的处理，从而降低系统耦合度，提高系统性能和可用性。

一般来说，可以抽离的任务具有以下的特点：

• 允许延后|异步|并行处理 （相对于传统的 即时|同步|串行 的执行方式）

• 允许延后：

  抢购活动时，先快速缓冲有限的参与人数到消息队列，后续再排队处理实际的抢购业务；

• 允许异步：

  业务处理过程中的邮件，短信等通知

• 允许并行：

  用户支付成功之后，邮件通知，微信通知，短信通知可以由多个不同的消费者并行执行，通知到达的时间不要求先后顺序。

• 允许失败和重试

• 强一致性的业务放入核心流程处理

• 无一致性要求或最终一致即可的业务放入队列处理

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

• 消息的发布，获取，执行，删除，重发，失败处理，延迟执行，超时控制等

• 队列的多队列， 内存限制 ，启动，停止，守护等

• 消息队列可降级为同步执行

thinkphp-queue 内置了 Redis，Database，Topthink ，Sync这四种驱动。本文主要介绍 thinkphp-queue 结合其内置的 redis 驱动的使用方式和基本原理。

注1：如无特殊说明，下文中的 ‘消息’ 和 ‘任务’两个词指代的是同一个概念，即队列中的一个成员。该成员对消息队列而言是其内部保存的消息； 对业务应用而言是一个待执行的任务。请根据语境区分。

安装

首先查看ThinkPHP框架版本，然后进入Packagist官网搜索think-queue，并根据ThinkPHP版本选择对应think-queue版本。

thinkphp-queue地址：https://packagist.org/packages/topthink/think-queue

本文采用的ThinkPHP的版本为5.0.23。

可直接使用Composer为当前项目安装think-queue消息队列插件

搭建消息队列的存储环境

不推荐使用数据库，如果使用Redis驱动，那么需要提前安装Redis服务以及PHP的Redis扩展。

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

消息的创建与推送

我们在控制器中执行测试代码，将数据推送到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 ，更合理的做法是先定义好消息名称与消费者类名的映射关系，然后由某个可以获取该映射关系的类来推送这个消息。这样，生产者只需要知道消息的名称，而无需指定哪个消费者类来处理。

消息的消费与删除

编写 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;
     }
 }

发布任务

在浏览器中访问 http://your.project.domain/index/job_test/actionWithHelloJob ,可以看到消息推送成功。

消息推送成功后可以用redis可视化工具查看redis数据进行验证

处理任务

切换到当前终端到项目根目录

$ php think queue:work --queue dismiss_job_queue

查看执行的结果 

至此，成功地使用thinkphp中的thinkphp-queue经历了一个消息的 创建 -> 推送 -> 消费 -> 删除 的基本流程。