安裝環(huán)境
Mac環(huán)境安裝Docker寡具，使用Docker安裝RabbitMQ
本教程中使用php-amqplib，并且使用Composer依賴管理：php-amqplib
項目中添加一個 composer.json文件：
{
"require": {
"php-amqplib/php-amqplib": "2.6.1"
}
}
Rabbit官方文檔參考
Docker安裝RabbitMQ
// 搜索RabbitMQ鏡像
docker search rabbitmq
// 獲取RabbitMQ鏡像
docker pull rabbitmq
// 運行RabbitMQ鏡像 | 5672 15672 為開放端口 | -v 掛載目錄方便代碼修改自動上傳 | -e RABBITMQ_DEFAULT_USER=root -e RABBITMQ_DEFAULT_PASS=123 設(shè)置后臺登錄賬號密碼
docker run -d -it --name my-rabbit -p 5672:5672 -p 15672:15672 -v /mnt:/home -e RABBITMQ_DEFAULT_USER=root -e RABBITMQ_DEFAULT_PASS=123 rabbitmq
啟動訪問 http://127.0.0.1:15672 地址訪問RabbitMQ管理后臺，然后發(fā)現(xiàn)嘎~打不開！结耀！需要執(zhí)行以下命令
// 查看Docker中RabbitMQ容器
docker ps -l
// 進入Docker中RabbitMQ容器 | 容器ID為 docker ps -l 結(jié)果中`CONTAINER ID`列
docker exec -it 容器ID /bin/bash
// 打開RabbitMQ后臺管理訪問設(shè)置
rabbitmqctl start_app
rabbitmq-plugins enable rabbitmq_management
rabbitmqctl stop
成功訪問！Ｇ复荨２右狻！
image
image
RabbitMQ安裝RabbitMQ延遲隊列插件
RabbitMQ官方提供了延遲隊列插件rabbitmq_delayed_message_exchange-3.9.0.ez, 下載前請確認自己的RabbitMQ版本扫沼，下載對應版本的插件出爹，選擇ez格式文件
image.png
將插件放入宿主機與Docker的掛載目錄中庄吼，最終目的就是將插件放入RabbitMQ鏡像的/plugins目錄中
// 回顧Docker啟動命令，Docker中宿主機/mnt目錄與RabbitMQ鏡像的/home掛載
docker run -d -it --name my-rabbit -p 5672:5672 -p 15672:15672 -v /mnt:/home -e RABBITMQ_DEFAULT_USER=root -e RABBITMQ_DEFAULT_PASS=123 rabbitmq
// 將下載的插件移動到/mnt目錄中
mv rabbitmq_delayed_message_exchange-3.9.0.ez /mnt
// 把rabbitmq_delayed_message_exchange-3.9.0.ez移動至/plugins中
docker exec -it 容器ID /bin/bash
cd /nmt
mv rabbitmq_delayed_message_exchange-3.9.0.ez/plugins
// 執(zhí)行啟用插件命令
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
exit
docker restart 容器ID
重新登錄管理后臺 严就，出現(xiàn)如圖所示总寻，即插件安裝成功
image.png
PHP項目代碼
RabbitMQ類
class RabbitMQController {
private $host = "127.0.0.1";
private $port = 5672;
private $user = "root";
private $psss = 123;
private $msg;
private $channel;
private $connection;
// 過期時間
const TIMEOUT_5_S = 5; // 5s
const TIMEOUT_10_S = 10; // 10s
private $exchange_logs = "logs";
private $exchange_direct = "direct";
private $exchange_delayed = "delayed";
private $queue_delayed = "delayedQueue";
CONST EXCHANGETYPE_FANOUT = "fanout";
CONST EXCHANGETYPE_DIRECT = "direct";
CONST EXCHANGETYPE_DELAYED = "x-delayed-message";
public function __construct($type = false) {
$this->connection = new AMQPStreamConnection($this->host,$this->port,$this->user ,$this->psss);
$this->channel = $this->connection->channel();
// 聲明Exchange
$this->channel->exchange_declare($this->exchange_delayed, self::EXCHANGETYPE_DELAYED, false, true, false,false,false,new AMQPTable(["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);
}
/**
* delay creat message
*/
public function createMessageDelay($msg,$time) {
$delayConfig = [
'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
'application_headers' => new AMQPTable(['x-delay' => $time * 1000])
];
$msg = new AMQPMessage($msg,$delayConfig);
return $msg;
}
/**
* delay send message
*/
public function sendDelay($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();
}
/**
* delay consum
*/
public function consumDelay(){
$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 ' [*] Waiting for logs. To exit press CTRL+C', "\n";
while (count($this->channel->callbacks)) {
$this->channel->wait();
}
$this->channel->close();
$this->connection->close();
}
測試類
public function actionSendDelay($msg,$time){
$Rabbit = new RabbitMQController("x-delayed-message");
$Rabbit->sendDelay($msg,$time);
}
public function actionConsumDelay(){
$Rabbit = new RabbitMQController("x-delayed-message");
$Rabbit->consumDelay();
}
image.png
image.png
image.png
