<?php
/**
* 文件路徑： \application\index\controller\JobTest.php
* 該控制器的業(yè)務(wù)代碼中借助了thinkphp-queue 庫(kù)，將一個(gè)消息推送到消息隊(duì)列
*/
namespace app\index\controller;
  use think\Exception;

  use think\Queue;

  class JobTest {
  /**
   * 一個(gè)使用了隊(duì)列的 action
   */
  public function actionWithHelloJob(){
      
      // 1.當(dāng)前任務(wù)將由哪個(gè)類來(lái)負(fù)責(zé)處理。 
      //   當(dāng)輪到該任務(wù)時(shí)，系統(tǒng)將生成一個(gè)該類的實(shí)例，并調(diào)用其 fire 方法
      $jobHandlerClassName  = 'app\index\job\Hello'; 
      
      // 2.當(dāng)前任務(wù)歸屬的隊(duì)列名稱，如果為新隊(duì)列，會(huì)自動(dòng)創(chuàng)建
      $jobQueueName  	  = "helloJobQueue"; 
      
      // 3.當(dāng)前任務(wù)所需的業(yè)務(wù)數(shù)據(jù) . 不能為 resource 類型，其他類型最終將轉(zhuǎn)化為json形式的字符串
      //   ( jobData 為對(duì)象時(shí)，存儲(chǔ)其public屬性的鍵值對(duì) )
      $jobData       	  = [ 'ts' => time(), 'bizId' => uniqid() , 'a' => 1 ] ;
      
      // 4.將該任務(wù)推送到消息隊(duì)列，等待對(duì)應(yīng)的消費(fèi)者去執(zhí)行
      $isPushed = Queue::push( $jobHandlerClassName , $jobData , $jobQueueName );	
      
      // database 驅(qū)動(dòng)時(shí)，返回值為 1|false  ;   redis 驅(qū)動(dòng)時(shí)，返回值為 隨機(jī)字符串|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.';
      }
  }
 }

 <?php
  /**
   * 文件路徑： \application\index\job\Hello.php
   * 這是一個(gè)消費(fèi)者類，用于處理 helloJobQueue 隊(duì)列中的任務(wù)
   */
  namespace app\index\job;

  use think\queue\Job;

  class Hello {
      
      /**
       * fire方法是消息隊(duì)列默認(rèn)調(diào)用的方法
       * @param Job            $job      當(dāng)前的任務(wù)對(duì)象
       * @param array|mixed    $data     發(fā)布任務(wù)時(shí)自定義的數(shù)據(jù)
       */
      public function fire(Job $job,$data)
      {
          // 有些消息在到達(dá)消費(fèi)者時(shí),可能已經(jīng)不再需要執(zhí)行了
          $isJobStillNeedToBeDone = $this->checkDatabaseToSeeIfJobNeedToBeDone($data);
          if(!$isJobStillNeedToBeDone){
              $job->delete();
              return;
          }
        
          $isJobDone = $this->doHelloJob($data);
        
          if ($isJobDone) {
              // 如果任務(wù)執(zhí)行成功， 記得刪除任務(wù)
              $job->delete();
              print("<info>Hello Job has been done and deleted"."</info>\n");
          }else{
              if ($job->attempts() > 3) {
                  //通過(guò)這個(gè)方法可以檢查這個(gè)任務(wù)已經(jīng)重試了幾次了
                  print("<warn>Hello Job has been retried more than 3 times!"."</warn>\n");
                  
  				        $job->delete();
                  
                  // 也可以重新發(fā)布這個(gè)任務(wù)
                  //print("<info>Hello Job will be availabe again after 2s."."</info>\n");
                  //$job->release(2); //$delay為延遲時(shí)間，表示該任務(wù)延遲2秒后再執(zhí)行
              }
          }
      }
      
      /**
       * 有些消息在到達(dá)消費(fèi)者時(shí),可能已經(jīng)不再需要執(zhí)行了
       * @param array|mixed    $data     發(fā)布任務(wù)時(shí)自定義的數(shù)據(jù)
       * @return boolean                 任務(wù)執(zhí)行的結(jié)果
       */
      private function checkDatabaseToSeeIfJobNeedToBeDone($data){
          return true;
      }

      /**
       * 根據(jù)消息中的數(shù)據(jù)進(jìn)行實(shí)際的業(yè)務(wù)處理...
       */
      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;
      }
  }

$ php think queue:work --queue dismiss_job_queue