您当前的位置：学无止境 > rabbitmq死信队列的简单运用网站首页学无止境

rabbitmq死信队列的简单运用

发布时间：2024-07-16 14:35:45编辑：三青查看次数：1904

注意：默认根目录已经安装composer下载安装rabbitmq（php-amqplib/php-amqplib）扩展

rabbitmq死信队列生产者

dead_pub.php

<?php
    require_once __DIR__ . '/vendor/autoload.php';
    use PhpAmqpLibConnectionAMQPStreamConnection;
    use PhpAmqpLibMessageAMQPMessage;
    use PhpAmqpLibWireAMQPTable;
    
    $dbName = 'sanqing';
    $dbPwd = '111111';
    $tableName = 'order';
    
    $connection = new AMQPStreamConnection('localhost', 5672, $dbName, $dbPwd, $tableName);
    $channel = $connection->channel();
    $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';
    $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,false);	
    $channel->queue_bind($dead_queue_name,$dead_exc_name,$dead_routing_key);

    $data = 'this is '.$routing_key.' dead message';
    $msg = new AMQPMessage($data,['delivery_mode'=>AMQPMessage::DELIVERY_MODE_PERSISTENT]);
    $channel->basic_publish($msg,$exc_name,$routing_key);
    $channel->close();
    $connection->close();

rabbitmq死信队列消费者

dead_work.php

<?php
    require_once __DIR__ . '/vendor/autoload.php';
    use PhpAmqpLibConnectionAMQPStreamConnection;
    
    $dbName = 'sanqing';
    $dbPwd = '111111';
    $tableName = 'order';
    
    $connection = new AMQPStreamConnection('localhost', 5672, $dbName, $dbPwd, $tableName);
    $channel = $connection->channel();
    $dead_exc_name = 'dead_exc_pay';
    $dead_routing_key = 'dead_route_pay';
    $dead_queue_name = 'dead_queue_pay';

    $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();

启动消费者和生产者

php dead_work.php

php dead_pub.php

因为消息ttl过期时间设置的是20秒，所以消费者在生产者启动20秒后才能获得数据