第PHP實(shí)現(xiàn)異步延遲消息隊(duì)列的方法詳解目錄一、前言二、場(chǎng)景三、安裝RabbitMQ延遲隊(duì)列插件四、在Laravel框架中進(jìn)行使用五、執(zhí)行生產(chǎn)消息和消費(fèi)消息

一、前言

需求：電商秒殺場(chǎng)景中，如果用戶下單10分鐘未支付，需要進(jìn)行庫(kù)存歸還

本篇是用PHP+Laravel+RabbitMQ來實(shí)現(xiàn)異步延遲消息隊(duì)列

二、場(chǎng)景

在電商項(xiàng)目中，當(dāng)我們下單之后，一般需要20分鐘之內(nèi)或者30分鐘之內(nèi)付款，否則訂單就會(huì)進(jìn)入異常處理邏輯中，被取消，那么進(jìn)入到異常處理邏輯中，就可以當(dāng)成是一個(gè)延遲隊(duì)列

公司的會(huì)議預(yù)定系統(tǒng)，在會(huì)議預(yù)定成功后，會(huì)在會(huì)議開始前半小時(shí)通知所有預(yù)定該會(huì)議的用戶

安全工單超過24小時(shí)未處理，則自動(dòng)拉企業(yè)微信群提醒相關(guān)責(zé)任人

用戶下單外賣以后，距離超時(shí)時(shí)間還有10分鐘時(shí)提醒外賣小哥即將超時(shí)

很多場(chǎng)景下我們都需要延遲隊(duì)列。

本文以RabbitMQ為例來和大家聊一聊延遲隊(duì)列的玩法。

使用RabbitMQ的rabbitmq_delayed_message_exchange插件來實(shí)現(xiàn)定時(shí)任務(wù)，這種方案較簡(jiǎn)單。

三、安裝RabbitMQ延遲隊(duì)列插件

官網(wǎng)插件下載地址

我這里直接下載了最新版本，你們根據(jù)自己的rabbitmq版本號(hào)進(jìn)行下載

把下載好的文件移動(dòng)到rabbitmq的插件plugins下，以我自己的Mac為例子，放到了如下路徑

然后執(zhí)行安裝插件指令，如下

rabbitmq-plugins

enable

rabbitmq_delayed_message_exchange

最后重啟rabbitmq服務(wù)，并刷新查看exchanges交換機(jī)有沒有該插件

如上圖則延遲消息隊(duì)列插件安裝完成

四、在Laravel框架中進(jìn)行使用

新建rabbitmq服務(wù)類，包含延遲消息隊(duì)列生產(chǎn)消息，和消費(fèi)消息，如下

代碼如下：

php

namespaceApp\Http\Controllers\Service;

usePhpAmqpLib\Connection\AMQPStreamConnection;

usePhpAmqpLib\Message\AMQPMessage;

usePhpAmqpLib\Wire\AMQPTable;

classRabbitmqServer

private$host="";

private$port=5672;

private$user="guest";

private$password="guest";

private$msg;

private$channel;

private$connection;

//過期時(shí)間

constTIMEOUT_5_S=5;//5s

constTIMEOUT_10_S=10;//10s

private$exchange_logs="logs";

private$exchange_direct="direct";

private$exchange_delayed="delayed";

private$queue_delayed="delayedQueue";

constEXCHANGETYPE_FANOUT="fanout";

constEXCHANGETYPE_DIRECT="direct";

constEXCHANGETYPE_DELAYED="x-delayed-message";

publicfunction__construct($type=false)

$this-connection=newAMQPStreamConnection($this-host,$this-port,$this-user,$this-password);

$this-channel=$this-connection-channel();

//聲明Exchange

$this-channel-exchange_declare($this-exchange_delayed,self::EXCHANGETYPE_DELAYED,false,true,false,false,false,newAMQPTable(["x-delayed-type"=self::EXCHANGETYPE_DIRECT]));

$this-channel-queue_declare($this-queue_delayed,false,true,false,false);

$this-channel-queue_bind($this-queue_delayed,$this-exchange_delayed,$this-queue_delayed);

*delaycreatmessage

publicfunctioncreateMessageDelay($msg,$time)

$delayConfig=[

'delivery_mode'=AMQPMessage::DELIVERY_MODE_PERSISTENT,

'application_headers'=newAMQPTable(['x-delay'=$time*1000])

$msg=newAMQPMessage($msg,$delayConfig);

return$msg;

*delaysendmessage

publicfunctionsendDelay($msg,$time=self::TIMEOUT_10_S)

$msg=$this-createMessageDelay($msg,$time);;

$this-channel-basic_publish($msg,$this-exchange_delayed,$this-queue_delayed);

$this-channel-close();

$this-connection-close();

*delayconsum

publicfunctionconsumDelay()

$callback=function($msg){

echo'[x]',$msg-body,"\n";

$this-channel-basic_ack($msg-delivery_info['delivery_tag'],false);

$this-channel-basic_qos(null,1,null);

$this-channel-basic_consume($this-queue_delayed,'',false,false,false,false,$callback);

echo'[*]Waitingforlogs.ToexitpressCTRL+C',"\n";

while(count($this-channel-callbacks)){

$this-channel-wait();

$this-channel-close();

$this-connection-close();

}

比如新建QueueController控制器，進(jìn)行測(cè)試生產(chǎn)消息放到延遲消息隊(duì)列中

代碼如下：

php

namespaceApp\Http\Controllers\Api\v1;

useApp\Http\Controllers\Controller;

useApp\Http\Controllers\Service\RabbitmqServer;

useApp\Jobs\Queue;

useIlluminate\Http\Request;

classQueueControllerextendsController

publicfunctionindex(Request$request)

//比如說現(xiàn)在是下訂單操作

//需求：如果用戶10分鐘之內(nèi)不支付訂單就要取消訂單，并且?guī)齑鏆w還

$msg=$request-post();

$Rabbit=newRabbitmqServer("x-delayed-message");

//第一個(gè)參數(shù)發(fā)送的消息，第二個(gè)參數(shù)延遲多少秒

$Rabbit-sendDelay(json_encode($msg),5);

}

至此通過接口調(diào)試工具進(jìn)行模擬生產(chǎn)消息即可

消息生產(chǎn)完畢要進(jìn)行消費(fèi)，這里使用的是Laravel的任務(wù)調(diào)度，代碼如下

php

namespaceApp\Console\Commands;

useApp\Http\Controllers\Service\RabbitmqServer;

useIlluminate\Console\Command;

usePhpAmqpLib\Connection\AMQPStreamConnection;

usePhpAmqpLib\Message\AMQPMessage;

classRabbitmqConsumerCommandextendsCommand

*Thenameandsignatureoftheconsolecommand.

*@varstring

protected$signature='rabbitmq_consumer';//給消費(fèi)者起個(gè)command名稱

*Theconsolecommanddescription.

*@varstring

protected$description='Commanddescription';

*Createanewcommandinstance.

*@returnvoid

publicfunction__construct()

parent::__construct();

*Executetheconsolecommand.