安装环境

Mac环境安装Docker，使用Docker安装RabbitMQ

本教程中使用php-amqplib，并且使用Composer依赖管理：php-amqplib
项目中添加一个 composer.json文件：

 {
     "require": {
      "php-amqplib/php-amqplib": "2.6.1"
     }
 }

Rabbit官方文档参考

AMQP 协议
RabbitMQ-PHP版本

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启动命令，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