目錄

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

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

前言

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

一般來(lái)說(shuō)，可以抽離的任務(wù)具有以下的特點(diǎn)：

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

  • 允許延后：

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

  • 允許異步：

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

  • 允許并行：

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

• 允許失敗和重試

  • 強(qiáng)一致性的業(yè)務(wù)放入核心流程處理
  • 無(wú)一致性要求或最終一致即可的業(yè)務(wù)放入隊(duì)列處理

thinkphp-queue 是thinkphp 官方提供的一個(gè)消息隊(duì)列服務(wù)，它支持消息隊(duì)列的一些基本特性：

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

thinkphp-queue 內(nèi)置了 Redis，Database，Topthink ，Sync這四種驅(qū)動(dòng)。本文主要介紹 thinkphp-queue 結(jié)合其內(nèi)置的 redis 驅(qū)動(dòng)的使用方式和基本原理。

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

安裝

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

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

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

可直接使用Composer為當(dāng)前項(xiàng)目安裝think-queue消息隊(duì)列插件

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

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

根據(jù)選擇的存儲(chǔ)方式，在 \application\config\queue.php 這個(gè)配置文件中，添加消息隊(duì)列對(duì)應(yīng)的驅(qū)動(dòng)配置

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

我們?cè)诳刂破髦袌?zhí)行測(cè)試代碼，將數(shù)據(jù)推送到helloJobQueue隊(duì)列

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

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

在這個(gè)例子當(dāng)中，我們是手動(dòng)指定的 $jobHandlerClassName ，更合理的做法是先定義好消息名稱與消費(fèi)者類名的映射關(guān)系，然后由某個(gè)可以獲取該映射關(guān)系的類來(lái)推送這個(gè)消息。這樣，生產(chǎn)者只需要知道消息的名稱，而無(wú)需指定哪個(gè)消費(fèi)者類來(lái)處理。

消息的消費(fèi)與刪除

編寫(xiě) Hello 消費(fèi)者類，用于處理 helloJobQueue 隊(duì)列中的任務(wù)

新增 \application\index\job\Hello.php 消費(fèi)者類，并編寫(xiě)其 fire() 方法

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

發(fā)布任務(wù)

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

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

處理任務(wù)

切換到當(dāng)前終端到項(xiàng)目根目錄

$ php think queue:work --queue dismiss_job_queue

查看執(zhí)行的結(jié)果 

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

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