**本文以thinkphp5中擴展 think-queue的應用為實例**
1.安裝think-queue 擴展
```
composer install topthink/think-queue
```
2.index模塊下新建控制器,調用消息隊列入隊
~~~
```
<?php
namespace app\index\controller;
/**
* 文件路徑： \application\index\controller\JobTest.php
* 該控制器的業務代碼中借助了thinkphp-queue 庫，將一個消息推送到消息隊列
*/
use think\Exception;
use think\Queue;
use think\Log;
class JobTest
{
/**
* 一個使用了隊列的 action
*/
public function actionWithHelloJob($msg){
// 1.當前任務將由哪個類來負責處理。
// 當輪到該任務時，系統將生成一個該類的實例，并調用其 fire 方法
$jobHandlerClassName = 'app\index\job\Hello'; // 對應job下處理的消息類
// 2.當前任務歸屬的隊列名稱，如果為新隊列，會自動創建
$jobQueueName = "helloJobQueue";
// 3.當前任務所需的業務數據 . 不能為 resource 類型，其他類型最終將轉化為json形式的字符串
// ( jobData 為對象時，存儲其public屬性的鍵值對 )
$jobData = [ 'ts' => time(), 'bizId' => uniqid() , 'a' => 1, 'msg'=>$msg , 'time'=>date('Y-m-d H:i:s') ] ;
// 4.將該任務推送到消息隊列，等待對應的消費者去執行
$isPushed = Queue::push( $jobHandlerClassName , $jobData , $jobQueueName );
// database 驅動時，返回值為 1|false ; redis 驅動時，返回值為 隨機字符串|false
if( $isPushed !== false ){
return date('Y-m-d H:i:s') . " a new Hello Job is Pushed to the MQ"."<br>";
}else{
return 'Oops, something went wrong.';
}
}
public function actionWithMultiTask($whichTask){
$taskType = $_GET['taskType'];
switch ($whichTask) {
case 'taskA':
$jobHandlerClassName = 'application\index\job\MultiTask@taskA';
$jobDataArr = ['a' => '1'];
$jobQueueName = "multiTaskJobQueue";
break;
case 'taskB':
$jobHandlerClassName = 'application\index\job\MultiTask@taskB';
$jobDataArr = ['b' => '2'];
$jobQueueName = "multiTaskJobQueue";
break;
default:
break;
}
$isPushed = Queue::push($jobHandlerClassName, $jobDataArr, $jobQueueName);
if ($isPushed !== false) {
echo("the $taskType of MultiTask Job has been Pushed to ".$jobQueueName ."<br>");
}else{
throw new Exception("push a new $taskType of MultiTask Job Failed!");
}
}
}
~~~
```
3.index模塊下新建job目錄,供調用處理進入消息隊列的消息
<?php
namespace app\index\job;
use think\Log;
use think\queue\Job;
/**
* 文件路徑： \application\index\job\Hello.php
* 這是一個消費者類，用于處理 helloJobQueue 隊列中的任務
*/
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) {
// 根據消息中的數據進行實際的業務處理...
$this->testWriteLog($data['msg'],$data['time']);
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;
}
/**
* 耗時方法,測試是否異步執行
*
*/
public function testWriteLog($msg,$time)
{
sleep(30);
Log::info('測試異步執行:'.$msg.'-'.$time.'-當前時間:'.date('Y-m-d H:i:s'));
}
}
```
4.windows下執行進入項目根目錄 php think queue:listen 監聽消息隊列發布進入
5.啟動 本地redis
6.訪問控制器,發布消息隊列任務
7.啟動消息隊列執行出 php think queue:work --queue helloJobQueue
