RabbitMQ在Tp5.0 中实现延时队列（死信队列）

实现步骤：
创建两个交换机和队列：死信队列和死信队列上游队列
死信队列的特性：
插入一个没有过期时间的消息会导致，队列永远不会被消费
前面的过期时间太长了，后面的消息就算过期也不会被消费的
按随机串的方式来生成交换机和队列，保证一个交换机只有一条消息，消费完以后删除掉
这里是设置消息的过期时间方式来实现的延时任务
只需要监听死信队列的上游队列就可以实现想要的效果
<?php
namespace app\index\controller;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
use think\Log;


class Index
{

    const consumerTag = 'consumer'; // 消费者标签
    const exchange = 'router'; // 交换机名
    const queue = 'msgs';// 的队列名

    /**
     * 推入消息到队列中
     */
    public static  function pushMessage()
    {

        // 连接rabbitMQ
        $connection = new AMQPStreamConnection('localhost', 5672, 'dome', 'dome', '/');
        // 开启一个信道
        $channel = $connection->channel();

        // 动态的创建交换机和队列
        // 以免在死信队列中出现时间太长的消息，造成队列阻塞
        // 过期时长（毫秒）
        $timeout =  10000;
        $rand_num = time() . rand(100000, 999999);
        $exchange_name =  "cache_exchange_" . $rand_num;
        $queue_name =  "queue_" . $rand_num;

        // exchange 交换机名称
        // type 交换器类型
        // passive 检测交换机是否存在   true 只检测不创建 false 创建
        // durable 是否持久化队列 true 为持久化
        // auto_delete  当所有绑定队列都不在使用时，是否自动删除交换器 true：删除false：不删除

        /**死信队列交换机*/
        $channel->exchange_declare($exchange_name, 'direct', false, false, false);
        /**死信队列上游交换机， 队列内的消息过期以后会把消息推到这个队列中*/
        $channel->exchange_declare('delay_exchange', 'direct',false,false,false);

        //  两个参数，来控制队列出现  dead letter 的时候，重新发送消息的目的地
        $tale = new AMQPTable();
        $tale->set('x-dead-letter-exchange', 'delay_exchange');//  表示过期后由哪个exchange处理
        $tale->set('x-dead-letter-routing-key', 'delay_exchange');
        $tale->set('x-message-ttl', $timeout);  //存活时长（毫秒）， 下面的过期时间不能超过


        // 绑定队列和交换机 --- 死信队列
        // queue 队列名
        // passive 检测队列是否存在  true 只检测不创建 false 创建
        // durable 是否持久化队列 true 为持久化
        // exclusive 私有队列 不允许其它用户访问  设置true 将会变成私有
        // auto_delete  当所有消费客户端连接断开后，是否自动删除队列
        $channel->queue_declare($queue_name, false,true,false,false,false, $tale);
        $channel->queue_bind($queue_name, $exchange_name, $exchange_name);

        // 绑定队列和交换机 --- 死信队列 上游队列
        // 消息过期会交给这个队列进行处理
        $channel->queue_declare('delay_queue', false, true, false, false);
        $channel->queue_bind('delay_queue', 'delay_exchange', 'delay_exchange');

        // 写入队列的消息
        $data['message'] = [
            'name' => '队列的消息体'
        ];
        $data['exchange'] = $exchange_name; // 死信队列交换机
        $data['queue'] = $queue_name; // 死信队列
        $messageBody = json_encode($data) ;

        // 消息内容
        // delivery_mode  投递模式  delivery mode 设置为 2标记持久化
        $message = new AMQPMessage(
            $messageBody,
            array(
                'expiration' => intval($timeout),
                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
            ));

        // $message 消息内容
        // $exchange  交换器名称
        // routing_key   路由键 (routing key)  主题交换机会用到
        $channel->basic_publish($message, $exchange_name, $exchange_name);
        // 关闭信道
        $channel->close();
        //关闭 amqp 连接
        $connection->close();
        return "ok";
    }


    function shutdown($channel, $connection)
    {
        $channel->close();
        $connection->close();
        Log::info("closed",3);
    }

    function process_message($message)
    {
        // *****这里写自己的逻辑，我这里测试方便写了一个获取消息写到日志中
        Log::info("error data111111111111111:" . json_encode($message) , 2);
        // *****这里写自己的逻辑，我这里测试方便写了一个获取消息写到日志中

        $body = $body = json_decode($message->body, true) ;
        $exchange_name = $body['exchange'];
        $queue_name = $body['queue'];

        // 一个队列只有一条消息，消费完就直接删除，交换机和队列
        $message->delivery_info['channel']->exchange_delete($exchange_name);
        $message->delivery_info['channel']->queue_delete($queue_name);

        $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
        // Send a message with the string "quit" to cancel the consumer.
        if ($message->body === 'quit') {
            $message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']);
        }
    }

    /**
     * 启动
     *
     * @return \think\Response
     */
    public function start()
    {

        $connection = new AMQPStreamConnection('localhost', 5672, 'dome', 'dome', '/');
        $channel = $connection->channel();
        $channel->queue_declare('delay_queue', false, true, false, false);
        $channel->exchange_declare('delay_exchange', 'direct', false, false, false);
        $channel->queue_bind('delay_queue', 'delay_exchange','delay_exchange');


        // queue   队列名称
        // consumer_tag  消费者标签
        // no_ack  在设置了 no_ack=false 的情况下）只要 consumer 手动应答了 Basic.Ack ，就算其“成功”处理了
        // no_ack=true （此时为自动应答）
        // exclusive  是否是私有队列 设置true 将会变成私有
        // callback = null, 回调函数
        $channel->basic_consume('delay_queue', self::consumerTag, false, false, false, false, array($this, 'process_message'));

        // 不管你的php代码执行是否成功，最后都会执行 shutdown方法，关闭信道和连接
        register_shutdown_function(array($this, 'shutdown'), $channel, $connection);
        while (count($channel->callbacks)) {
            $channel->wait();
        }
        Log::info ("starting",3);
    }

}