您当前的位置：学无止境 > rabbitmq的direct模式消息队列的简单运用网站首页学无止境

rabbitmq的direct模式消息队列的简单运用

发布时间：2024-07-16 14:14:21编辑：三青查看次数：1832

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

direct模式消息队列生产者

public.php（$routing_key的不同，调用不同的消费者）

<?php
    require_once __DIR__ . '/vendor/autoload.php';
    use PhpAmqpLibConnectionAMQPStreamConnection;
    use PhpAmqpLibMessageAMQPMessage;
    
    $dbName = 'sanqing';
    $dbPwd = '111111';
    $tableName = 'order';
    
    $connection = new AMQPStreamConnection('localhost', 5672, $dbName, $dbPwd, $tableName);
    $channel = $connection->channel();
    $exc_name = 'direct_log';
    $routing_key = 'info';
    $channel->exchange_declare($exc_name,'direct',false,false,false);
    $data = 'this is '.$routing_key.' direct message';
    $msg = new AMQPMessage($data,['delivery_mode'=>AMQPMessage::DELIVERY_MODE_PERSISTENT]);
    $channel->basic_publish($msg,$exc_name,$routing_key);
    $channel->close();
    $connection->close();

direct模式消息队列消费者

info.php（$routing_key的不同，调用不同的消费者）

<?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();
    $exc_name = 'direct_log';
    $routing_key = 'info';
    $channel->exchange_declare($exc_name,'direct',false,false,false);
    list($queue_name,,) = $channel->queue_declare('',false,false,true,false);    
    $channel->queue_bind($quene_name,$exc_name,$routing_key);
    $callback = function($msg){
        echo 'received '.$msg->body."n";
        $msg->ack();
    };
    $channel->basic_qos(null,1,null);
    $channel->basic_consume($queue_name,'',false,false,false,false,$callback);
    while($channel->is_open()){
    	$channel->wait();
    }
    
    $channel->close();
    $connection->close();

error.php（$routing_key的不同，调用不同的消费者）


<?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();
    $exc_name = 'direct_log';
    $routing_key = 'error';
    $channel->exchange_declare($exc_name,'direct',false,false,false);
    list($queue_name,,) = $channel->queue_declare('',false,false,true,false);    
    $channel->queue_bind($quene_name,$exc_name,$routing_key);
    $callback = function($msg){
        echo 'received '.$msg->body."n";
        $msg->ack();
    };
    $channel->basic_qos(null,1,null);
    $channel->basic_consume($queue_name,'',false,false,false,false,$callback);
    while($channel->is_open()){
    	$channel->wait();
    }
    
    $channel->close();
    $connection->close();