目錄

• 前言
• 安裝
• 搭建消息隊列的存儲環境
• 消息的創建與推送
• 消息的消費與刪除
• 發布任務
• 處理任務

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經歷了一個消息的 創建 -> 推送 -> 消費 -> 刪除 的基本流程。

到此這篇關于ThinkPHP基于think-queue的隊列插件實現消息推送的文章就介紹到這了,更多相關think-queue消息推送內容請搜索以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持！