[PHP知识库]php 使用MQ实现延迟队列

MQ延迟队列插件安装?

linux下mq安装与下载

生产者

 public function send_msg($id){
        $name = 'dead-x-tp6';//交换机名
        $rounting_key = 'dead-x-key_tp6'; // 交换机路由key
        $queue = 'dead-x-queue_tp6'; // 队列名
        $ttl = 20000; // 过期时间毫秒

        $connection = new AMQPStreamConnection('localhost', 5672, 'admin', '123456','order');
        $channel = $connection->channel(); // 创建频道

        $channel->exchange_declare($name,'x-delayed-message', false, true, false);//申明交换机

        $args = new AMQPTable(['x-delayed-type' => 'direct']);

        $channel->queue_declare($queue, false, true, false, false, false, $args); // 申明队列
        $channel->queue_bind($queue, $name, $rounting_key);  // 队列绑定

        $data = [
            'key' => $rounting_key,
            'id' => $id
        ];

        $options = [
            'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,   //消息持久化
            'application_headers' => new AMQPTable(['x-delay' => $ttl])
        ];
        $msg = new  AMQPMessage(json_encode($data), $options);
        $channel->basic_publish($msg, $name, $rounting_key);
        $channel->close();
        $connection->close();
    }

消费者：

$name = 'dead-x-tp6';//交换机名
        $rounting_key = 'dead-x-key_tp6'; // 交换机路由key
        $queue = 'dead-x-queue_tp6'; // 队列名
        $connection = new AMQPStreamConnection('localhost', 5672, 'admin', '123456','order');
        $channel = $connection->channel(); // 创建频道
        $channel->exchange_declare($name,'x-delayed-message', false, true, false);

        $channel->queue_bind($queue, $name, $rounting_key); // 队列与交换机绑定

        $callback = function ($msg) {
            $data = json_decode($msg->body,true);
            $id = $data['id'];
            $status = Db::name('test')->where(['id' => $id])->value('status');
            if ( $status == 0) {
                Db::name('test')->where(['id' => $id])->update(['status' => 200]);
            }
            $msg->ack();
        };

        $channel->basic_qos(null,1,null);//只有consumer已经处理并确认了上一条message时queue才分派新的message给它
        $channel->basic_consume($queue, '', false, false, false, false, $callback);

        while(count($channel->callbacks))
        {
            $channel->wait();
        }

        $channel->close();
        $connection->close();
        // 指令输出
        $output->writeln('dead_msg');