当前位置：首页 > news >正文

RabbitMQ-消息队列延迟队列一

news 2025/12/3 12:28:42

1、安装rabbitmq

怎么安装rabbitmq请查看之前课程，如果已经安装，请略过此步。

2、创建vendor文件夹或是直接采用PHP框架

mkdir vendor

3、进入文件

cd vendor

4、安装php扩展

composer require php-amqplib/php-amqplib

5、进入上级创建dead文件夹

cd ../
mkdir dead

6、进入dead文件并创建生产者php

cd dead
touch dead_pub.php

7、输入dead生产者内容

<?phprequire_once  '../vendor/autoload.php';use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;$v_host = 'order';$exc_name ='exc_pay';
$routing_key = 'route_pay';
$queue_name = 'queue_pay';
$ttl = 20000;$dead_exc_name = 'dead_exc_pay';
$dead_routing_key = 'dead_route_pay';
$dead_queue_name = 'dead_queue_pay';$connection = new AMQPStreamConnection('localhost',5672,'wangyulong','wangyulong',$v_host);$channel = $connection->channel();$channel->exchange_declare($exc_name,'direct',false,false,false);$args = new AMQPTable(['x-message-ttl'=>$ttl,'x-dead-letter-exchange'=>$dead_exc_name,'x-dead-letter-routing-key'=>$dead_routing_key]);$channel->queue_declare($queue_name,false,true,false,false,false,$args);$channel->queue_bind($queue_name,$exc_name,$routing_key);//声明死信交换器  队列$channel->exchange_declare($dead_exc_name,'direct',false,false,false);$channel->queue_declare($dead_queue_name,false,true,false,false);$channel->queue_bind($dead_queue_name,$dead_exc_name ,$dead_routing_key);$data = 'this is dead message';$msg = new AMQPMessage($data,['delivery_mode'=>AMQPMEssage::DELIVERY_MODE_PERSISTENT]);$channel->basic_publish($msg,$exc_name,$routing_key);$channel->close();$connection->close();

8、创建消费者php

touch dead_worker.php

9、输入dead消费者内容

<?phprequire_once  '../vendor/autoload.php';use PhpAmqpLib\Connection\AMQPStreamConnection;$v_host ='order';$dead_exc_name = 'dead_exc_pay';
$dead_routing_key = 'dead_route_pay';
$dead_queue_name = 'dead_queue_pay';$connection = new AMQPStreamConnection('localhost',5672,'wangyulong','wangyulong',$v_host);$channel = $connection->channel();$channel->exchange_declare($dead_exc_name,'direct',false,false,false);$channel->queue_bind($dead_queue_name,$dead_exc_name,$dead_routing_key);$callback = function($msg){echo 'received ' ,$msg->body,"\n";$msg->ack();
};$channel->basic_qos(null,1,null);$channel->basic_consume($dead_queue_name,'',false,false,false,false,$callback);while($channel->is_open()){$channel->wait();
}$channel->close();$connection->close();