ThinkPHP基於think-queue的隊列插件實現消息推送

<?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.';
      }
  }
 }

 <?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;
      }
  }

$ php think queue:work --queue dismiss_job_queue