$queue = new AMQPQueue($channel);
$queue->setName($params['queueName']?:'');
$queue->setFlags(AMQP_DURABLE);
$queue->setArguments(array(
        'x-dead-letter-exchange' => 'delay_exchange',
        'x-dead-letter-routing-key' => 'delay_route',
        'x-message-ttl' => 60000,
));
$queue->declareQueue();

//来源公众号：【码农编程进阶笔记】
//header('Content-Type:text/html;charset=utf8;');

$params = array(
    'exchangeName' => 'delay_exchange',
    'queueName' => 'delay_queue',
    'routeKey' => 'delay_route',
);

$connectConfig = array(
    'host' => 'localhost',
    'port' => 5672,
    'login' => 'guest',
    'password' => 'guest',
    'vhost' => '/'
);

//var_dump(extension_loaded('amqp'));
try {
    $conn = new AMQPConnection($connectConfig);
    $conn->connect();
    if (!$conn->isConnected()) {
        //die('Conexiune esuata');
        //TODO 记录日志
        echo 'rabbit-mq 连接错误:', json_encode($connectConfig);
        exit();
    }
    $channel = new AMQPChannel($conn);
    if (!$channel->isConnected()) {
        // die('Connection through channel failed');
        //TODO 记录日志
        echo 'rabbit-mq Connection through channel failed:', json_encode($connectConfig);
        exit();
    }
    $exchange = new AMQPExchange($channel);
    $exchange->setFlags(AMQP_DURABLE);//声明一个已存在的交换器的，如果不存在将抛出异常，这个一般用在consume端
    $exchange->setName($params['exchangeName']?:'');
    $exchange->setType(AMQP_EX_TYPE_DIRECT); //direct类型
    $exchange->declareExchange();

    //$channel->startTransaction();

    $queue = new AMQPQueue($channel);
    $queue->setName($params['queueName']?:'');
    $queue->setFlags(AMQP_DURABLE);
    $queue->declareQueue();

    //绑定
    $queue->bind($params['exchangeName'], $params['routeKey']);
} catch(Exception $e) {
    echo $e->getMessage();
    exit();
}

function callback(AMQPEnvelope $message) {
    global $queue;
    if ($message) {
        $body = $message->getBody();
        echo '接收时间：'.date("Y-m-d H:i:s", time()). PHP_EOL;
        echo '接收内容：'.$body . PHP_EOL;
        //为了防止接收端在处理消息时down掉，只有在消息处理完成后才发送ack消息
        $queue->ack($message->getDeliveryTag());
    } else {
        echo 'no message' . PHP_EOL;
    }
}

//$queue->consume('callback');  第一种消费方式,但是会阻塞,程序一直会卡在此处

//注意：这里需要注意的是这个方法：$queue->consume，queue对象有两个方法可用于取消息：consume和get。前者是阻塞的，无消息时会被挂起，适合循环中使用；后者则是非阻塞的，取消息时有则取，无则返回false。
//就是说用了consume之后，会同步阻塞，该程序常驻内存，不能用nginx，apache调用。 

$action = '2';
if($action == '1'){
    $queue->consume('callback');  //第一种消费方式,但是会阻塞,程序一直会卡在此处
}else{
    //第二种消费方式,非阻塞
    $start = time();
    while(true)
    {
        $message = $queue->get();
        if(!empty($message))
        {
            echo '接收时间：'.date("Y-m-d H:i:s", time()). PHP_EOL;
            echo '接收内容：'.$message->getBody().PHP_EOL;
            $queue->ack($message->getDeliveryTag());    //应答，代表该消息已经消费
            $end = time();
            echo '运行时间：'.($end - $start).'秒'.PHP_EOL;
            //exit();
        }
        else
        {
            //echo 'message not found' . PHP_EOL;
        }
    }
}

//来源公众号：【码农编程进阶笔记】
//header('Content-Type:text/html;charset=utf-8;');

$params = array(
    'exchangeName' => 'test_cache_exchange',
    'queueName' => 'test_cache_queue',
    'routeKey' => 'test_cache_route',
);

$connectConfig = array(
    'host' => 'localhost',
    'port' => 5672,
    'login' => 'guest',
    'password' => 'guest',
    'vhost' => '/'
);

//var_dump(extension_loaded('amqp')); 判断是否加载amqp扩展
//exit();
for($i=5;$i>0;$i--){

    try {
        $conn = new AMQPConnection($connectConfig);
        $conn->connect();
        if (!$conn->isConnected()) {
            //die('Conexiune esuata');
            //TODO 记录日志
            echo 'rabbit-mq 连接错误:', json_encode($connectConfig);
            exit();
        }
        $channel = new AMQPChannel($conn);
        if (!$channel->isConnected()) {
            // die('Connection through channel failed');
            //TODO 记录日志
            echo 'rabbit-mq Connection through channel failed:', json_encode($connectConfig);
            exit();
        }
        $exchange = new AMQPExchange($channel);
        $exchange->setFlags(AMQP_DURABLE);//持久化
        $exchange->setName($params['exchangeName']);
        $exchange->setType(AMQP_EX_TYPE_DIRECT); //direct类型
        $exchange->declareExchange();

        //$channel->startTransaction();
        //RabbitMQ不容许声明2个相同名称、配置不同的Queue队列
        $queue = new AMQPQueue($channel);
        $queue->setName($params['queueName'].$i);
        $queue->setFlags(AMQP_DURABLE);
        $queue->setArguments(array(
            'x-dead-letter-exchange' => 'delay_exchange',       //// 死信交换机
            'x-dead-letter-routing-key' => 'delay_route',          // 死信路由
            'x-message-ttl' => (10000*$i),       // 当上面的消息扔到该队列中后，过了60秒，如果没有被消费，它就死了
            // 在RMQ中想要使用优先级特性需要的版本为3.5+。
            //'x-max-priority'=>0,//将队列声明为优先级队列，即在创建队列的时候添加参数 x-max-priority 以指定最大的优先级，值为0-255（整数）。
        ));
        $queue->declareQueue();

        //绑定队列和交换机
        $queue->bind($params['exchangeName'], $params['routeKey'].$i);
        //$channel->commitTransaction();
    } catch(Exception $e) {

    }

    // 当mandatory标志位设置为true时，如果exchange根据自身类型和消息routeKey无法找到一个符合条件的queue，那么会调用basic.return方法将消息返还给生产者；当mandatory设为false时，出现上述情形broker会直接将消息扔掉。
    //delivery_mode=2指明message为持久的
    //生成消息
    echo '发送时间：'.date("Y-m-d H:i:s", time()).PHP_EOL;
    echo 'i='.$i.'，延迟'.($i*10).'秒'.PHP_EOL;
    $message = json_encode(['order_id'=>time(),'i'=>$i]);
    $exchange->publish($message, $params['routeKey'].$i, AMQP_MANDATORY, array('delivery_mode'=>2));
    $conn->disconnect();
    sleep(2);
}

  $queue->setArguments(array(
            'x-dead-letter-exchange' => 'delay_exchange',       // 死信交换机
            'x-dead-letter-routing-key' => 'delay_route',          // 死信路由
            'x-message-ttl' => 10000,       // 当上面的消息扔到该队列中后，过了60秒，如果没有被消费，它就死了
        ));