安装环境
Mac环境安装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 设置后台登录账号密码
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管理后台，然后发现嘎~打不开！！需要执行以下命令
// 查看Docker中RabbitMQ容器
docker ps -l
// 进入Docker中RabbitMQ容器 | 容器ID为 docker ps -l 结果中`CONTAINER ID`列
docker exec -it 容器ID /bin/bash
// 打开RabbitMQ后台管理访问设置
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
// 执行启用插件命令
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
exit
docker restart 容器ID
重新登录管理后台 ，出现如图所示，即插件安装成功
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
