rabbitmq死信队列(延迟队列)demo

<?php

require_once './vendor/autoload.php';
use PhpAmqpLibConnectionAMQPStreamConnection;
use PhpAmqpLibMessageAMQPMessage;
use PhpAmqpLibWireAMQPTable;
use PhpAmqpLibExchangeAMQPExchangeType;

$connection = new AMQPStreamConnection('127.0.0.1', 5672, 'mq', 'mq123', '/');
$channel = $connection->channel();

$channel->exchange_declare('exchange.dlx', AMQPExchangeType::DIRECT, false, true);
$channel->exchange_declare('exchange.normal', AMQPExchangeType::FANOUT, false, true);
$args = new AMQPTable();
// 消息过期方式：设置 queue.normal 队列中的消息10s之后过期
$args->set('x-message-ttl', 10000);
$args->set('x-dead-letter-exchange', 'exchange.dlx');
$args->set('x-dead-letter-routing-key', 'routingkey');
$channel->queue_declare('queue.normal', false, true, false, false, false, $args);
$channel->queue_declare('queue.dlx', false, true, false, false);

$channel->queue_bind('queue.normal', 'exchange.normal');
$channel->queue_bind('queue.dlx', 'exchange.dlx', 'routingkey');
$message = new AMQPMessage('死信队列消息');
$channel->basic_publish($message, 'exchange.normal', 'rk');

$channel->close();
$connection->close();
————————————————

原文链接：https://blog.csdn.net/qq_42724459/article/details/109571547
生产者与消费者代码如下：

    /**
     * 入消息队列
     *
     * @param $queue string 队列名
     * @param $data mixed 数据
     */
    public  function pushMessageQueue( $data = null)
    {
        $host = config("queue.connections.rabbitmq.host");
        $port = config("queue.connections.rabbitmq.port");
        $login = config("queue.connections.rabbitmq.login");
        $password= config("queue.connections.rabbitmq.password");
        $vhost = config("queue.connections.rabbitmq.vhost");
        $exchange= $this->config["dds-zl-mq"]["exchange_name"];
        $queueName= $this->config["dds-zl-mq"]["queue_name"];
        $routingKey = $this->config["dds-zl-mq"]["router_key_name"]; //路由关键字(也可以省略)
        
        //建立生产者与mq之间的连接
        $connection = new AMQPStreamConnection($host, $port, $login, $password, $vhost);
        $channel = $connection->channel();//在已连接基础上建立生产者与mq之间的通道
        
        $channel->exchange_declare($exchange, 'direct', false, true, false); //声明初始化交换机
        $channel->queue_declare($queueName, false, true, false, false);//声明初始化一条队列
        $channel->queue_bind($queueName, $exchange, $routingKey); //将队列与某个交换机进行绑定，并使用路由关键字
        
        $msgBody = serialize($data); //json_encode($data);
        $msg = new AMQPMessage($msgBody, ['content_type' => 'text/plain', 'delivery_mode' => 2]); //生成消息
        $channel->basic_publish($msg, $exchange, $routingKey);//推送消息到某个交换机
        
        echo " [x] Sent 'Hello World!'
";
        
        $channel->close();
        $connection->close();
    }
    
    public function receive(){
        $host = config("queue.connections.rabbitmq.host");
        $port = config("queue.connections.rabbitmq.port");
        $login = config("queue.connections.rabbitmq.login");
        $password= config("queue.connections.rabbitmq.password");
        $vhost = config("queue.connections.rabbitmq.vhost");
        $exchange= $this->config["dds-zl-mq"]["exchange_name"];
        $queueName= $this->config["dds-zl-mq"]["queue_name"];
        $routingKey = $this->config["dds-zl-mq"]["router_key_name"]; //路由关键字(也可以省略)
        
        $connection = new AMQPStreamConnection($host, $port, $login, $password, $vhost);
        $channel = $connection->channel();
        
        $channel->exchange_declare($exchange, 'direct', false, true, false); //声明初始化交换机
        $channel->queue_declare($queueName, false, true, false, false);
        $channel->queue_bind($queueName, $exchange, $routingKey); //将队列与某个交换机进行绑定，并使用路由关键字
        
        echo " [*] Waiting for messages. To exit press CTRL+C
";
        
        $callback = function ($msg) {
            echo ' [x] Received ', $msg->body, "
";
        };
        
        $channel->basic_consume($queueName, '', false, true, false, false, $callback);
        
        while ($channel->is_consuming()) {//这个是阻塞模式，有消息就执行回调
            $channel->wait();
        }
    }