目錄

• 前言
• 安裝
• 搭建消息隊列的存儲環(huán)境
• 消息的創(chuàng)建與推送
• 消息的消費與刪除
• 發(fā)布任務
• 處理任務

think-queue是ThinkPHP官方提供的一個消息隊列服務，是專門支持隊列服務的擴展包。think-queue消息隊列適用于大并發(fā)或返回結果時間比較長且需要批量操作的第三方接口，可用于短信發(fā)送、郵件發(fā)送、APP推送。think-queue消息隊列可進行發(fā)布、獲取、執(zhí)行、刪除、重發(fā)、失敗處理、延遲執(zhí)行、超時控制等操作。

前言

傳統(tǒng)的程序執(zhí)行流程一般是 即時|同步|串行的，在某些場景下，會存在并發(fā)低，吞吐量低，響應時間長等問題。在大型系統(tǒng)中，一般會引入消息隊列的組件，將流程中部分任務抽離出來放入消息隊列，并由專門的消費者作針對性的處理，從而降低系統(tǒng)耦合度，提高系統(tǒng)性能和可用性。

一般來說，可以抽離的任務具有以下的特點：

• 允許延后|異步|并行處理 （相對于傳統(tǒng)的 即時|同步|串行 的執(zhí)行方式）

  • 允許延后：

    搶購活動時，先快速緩沖有限的參與人數(shù)到消息隊列，后續(xù)再排隊處理實際的搶購業(yè)務；

  • 允許異步：

    業(yè)務處理過程中的郵件，短信等通知

  • 允許并行：

    用戶支付成功之后，郵件通知，微信通知，短信通知可以由多個不同的消費者并行執(zhí)行，通知到達的時間不要求先后順序。

• 允許失敗和重試

  • 強一致性的業(yè)務放入核心流程處理
  • 無一致性要求或最終一致即可的業(yè)務放入隊列處理

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

• 消息的發(fā)布，獲取，執(zhí)行，刪除，重發(fā)，失敗處理，延遲執(zhí)行，超時控制等
• 隊列的多隊列， 內存限制 ，啟動，停止，守護等
• 消息隊列可降級為同步執(zhí)行

thinkphp-queue 內置了 Redis，Database，Topthink ，Sync這四種驅動。本文主要介紹 thinkphp-queue 結合其內置的 redis 驅動的使用方式和基本原理。

注1：如無特殊說明，下文中的 ‘消息’ 和 ‘任務’兩個詞指代的是同一個概念，即隊列中的一個成員。該成員對消息隊列而言是其內部保存的消息； 對業(yè)務應用而言是一個待執(zhí)行的任務。請根據(jù)語境區(qū)分。

安裝

首先查看ThinkPHP框架版本，然后進入Packagist官網搜索think-queue，并根據(jù)ThinkPHP版本選擇對應think-queue版本。

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

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

可直接使用Composer為當前項目安裝think-queue消息隊列插件

搭建消息隊列的存儲環(huán)境

不推薦使用數(shù)據(jù)庫，如果使用Redis驅動，那么需要提前安裝Redis服務以及PHP的Redis擴展。

根據(jù)選擇的存儲方式，在 \application\config\queue.php 這個配置文件中，添加消息隊列對應的驅動配置

消息的創(chuàng)建與推送

我們在控制器中執(zhí)行測試代碼，將數(shù)據(jù)推送到helloJobQueue隊列

新增 \application\index\controller\JobTest.php 控制器，在該控制器中添加 actionWithHelloJob 方法

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

  use think\Queue;

  class JobTest {
  /**
   * 一個使用了隊列的 action
   */
  public function actionWithHelloJob(){
      
      // 1.當前任務將由哪個類來負責處理。 
      //   當輪到該任務時，系統(tǒng)將生成一個該類的實例，并調用其 fire 方法
      $jobHandlerClassName  = "app\index\job\Hello"; 
      
      // 2.當前任務歸屬的隊列名稱，如果為新隊列，會自動創(chuàng)建
      $jobQueueName  	  = "helloJobQueue"; 
      
      // 3.當前任務所需的業(yè)務數(shù)據(jù) . 不能為 resource 類型，其他類型最終將轉化為json形式的字符串
      //   ( jobData 為對象時，存儲其public屬性的鍵值對 )
      $jobData       	  = [ "ts" => time(), "bizId" => uniqid() , "a" => 1 ] ;
      
      // 4.將該任務推送到消息隊列，等待對應的消費者去執(zhí)行
      $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     發(fā)布任務時自定義的數(shù)據(jù)
       */
      public function fire(Job $job,$data)
      {
  // 有些消息在到達消費者時,可能已經不再需要執(zhí)行了
  $isJobStillNeedToBeDone = $this->checkDatabaseToSeeIfJobNeedToBeDone($data);
  if(!$isJobStillNeedToBeDone){
      $job->delete();
      return;
  }

  $isJobDone = $this->doHelloJob($data);

  if ($isJobDone) {
      // 如果任務執(zhí)行成功， 記得刪除任務
      $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();
  
  // 也可以重新發(fā)布這個任務
  //print("<info>Hello Job will be availabe again after 2s."."</info>\n");
  //$job->release(2); //$delay為延遲時間，表示該任務延遲2秒后再執(zhí)行
      }
  }
      }
      
      /**
       * 有些消息在到達消費者時,可能已經不再需要執(zhí)行了
       * @param array|mixed    $data     發(fā)布任務時自定義的數(shù)據(jù)
       * @return boolean 任務執(zhí)行的結果
       */
      private function checkDatabaseToSeeIfJobNeedToBeDone($data){
  return true;
      }

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

發(fā)布任務

在瀏覽器中訪問 http://your.project.domain/index/job_test/actionWithHelloJob ,可以看到消息推送成功。

消息推送成功后可以用redis可視化工具查看redis數(shù)據(jù)進行驗證

處理任務

切換到當前終端到項目根目錄

$ php think queue:work --queue dismiss_job_queue

查看執(zhí)行的結果 

至此，成功地使用thinkphp中的thinkphp-queue經歷了一個消息的 創(chuàng)建 -> 推送 -> 消費 -> 刪除 的基本流程。

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