ThinkPHP5使用think-queue队列

ThinkPHP5怎么使用think-queue队列？

所使用的ThinkPHP版本为5.0，所以think-queue也只能用低版本的。

使用composer安装依赖：

composer require topthink/think-queue=1.1.6

在application/extra下创建队列配置文件queue.php，内容如下

<?php
use think\Env;

return [
    'connector' => 'Redis',            // Redis 驱动
    'expire' => null,                // 任务的过期时间，默认为60秒; 若要禁用，则设置为 null
    'default' => 'default',        // 默认的队列名称
    'host' =>  Env::get('redis.host','127.0.0.1'),        // redis 主机ip
    'port' => Env::get('redis.port', 6379),            // redis 端口
    'password' => Env::get('redis.password', ''),     // redis 密码
    'select' => Env::get('redis.select', 0),      // 使用哪一个 db，默认为 db0
    'timeout' => 0,                // redis连接的超时时间
    'persistent' => false,            // 是否是长连接

];

创建任务并推送到队列

import think\Queue;

public function queue_test()
    {
        // 1.当前任务将由哪个类来负责处理。
        //   当轮到该任务时，系统将生成一个该类的实例，并调用其 fire 方法
        $jobHandlerClassName = 'app\common\library\QueueTest';
        // 2.当前任务归属的队列名称，如果为新队列，会自动创建
        $jobQueueName = "testQueue";
        // 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.';
        }
    }

在application/common/library下创建QueueTest.php，用于消费队列

<?php

  namespace app\common\library;

  use think\queue\Job;

  class QueueTest {
      /**
       * 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->run($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;
      }
      
      /**
       * 根据消息中的数据进行实际的业务处理
       * @param array|mixed    $data     发布任务时自定义的数据
       * @return boolean                 任务执行的结果
       */

      private function run($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;
      }
  }

监听并执行队列

php think queue:work --queue testQueue --daemon --timeout 6000

该方法同样适用于以ThinkPHP5二开的框架，比如FastAdmin。