RabbitMQ学习之（五）_一个基于PHP的RabbitMQ操作类详解程序员

//amqp.php类文件 
<?php 
class Amqp 
{ 
public $e_name; 
public $q_name; 
public $k_route; 
public $channel; 
public function __construct($config,$e_name,$q_name,$k_route) 
{ 
$this->e_name = $e_name; 
$this->q_name = $q_name; 
$this->k_route = $k_route; 
//创建连接和channel  
$this->conn = new AMQPConnection($config);     
if (!$this->conn->connect()) {     
return array('error_code' => 1,'msg'=>'Cannot connect to the broker!' ); 
} 
$this->channel = new AMQPChannel($this->conn); 
$this->CreateExchange(); 
$this->CreateQueue(); 
} 
//创建交换机 
public function CreateExchange() 
{ 
$ex = new AMQPExchange($this->channel);     
$ex->setName($this->e_name);   
$ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型    
$ex->setFlags(AMQP_DURABLE | AMQP_AUTODELETE); //持久化   
//echo "Exchange Status:".$ex->declare()."/n";   //队列内容总数 
$ex->declare(); 
$this->ex = $ex; 
} 
//创建队列    
public function CreateQueue() 
{ 
$q = new AMQPQueue($this->channel);   
$q->setName($this->q_name);     
$q->setFlags(AMQP_DURABLE | AMQP_AUTODELETE); //持久化    
//echo "Message Total:".$this->q->declare()."/n";   
//绑定交换机与队列，并指定路由键   
//echo "queue status: ".$q->declare(); 
//echo "/n"; 
//echo 'Queue Bind: '.$q->bind($this->e_name, $this->k_route)."/n";  
//echo "/n"; 
          
$q->bind($this->e_name, $this->k_route); 
} 
//发送消息 
public function send($msg) 
{  
//$this->CreateExchange(); 
//$this->CreateQueue(); 
$message=json_encode($msg); 
$this->channel->startTransaction(); 
//echo "send: ".$this->ex->publish($message, $this->k_route); //将你的消息通过制定routingKey发送 
$status = $this->ex->publish($message, $this->k_route); 
$this->channel->commitTransaction(); 
$this->conn->disconnect(); 
return array('status'=>$status); 
} 
//获取消息 
public function get() 
{ 
$q = new AMQPQueue($this->channel);   
$q->setName($this->q_name);   
$q->setFlags(AMQP_DURABLE | AMQP_AUTODELETE);   
//$q->delete();删除队列 
$return=array(); 
while($a=$q->declare()) 
{ 
//echo "queue status: ".$a; 
//echo "==========/n"; 
 
$messages = $q->get(AMQP_AUTOACK); 
$return[]=json_decode($messages->getBody(),true);   
//echo "/n";   
        } 
$this->conn->disconnect(); 
return $return; 
} 
}

//config.php配置文件 
return array( 
'amqp'=>array(     
array(     
'host' => '127.0.0.1', 
'port' => '5672', 
'vhost' => '/', 
'user' => 'admin', 
'password' => 'admin' 
) 
), 
);

//send.php(加入队列文件|生产者) 
require_once('amqp.php'); 
$e_name = 'e_guest'; //交换机名   
$k_route = 'k_route_feedpush'; //路由key 
$q_name = 'q_guest_feedpush'; //队列名 
$config = config('amqp'); 
$amqp = new Amqp(config('amqp'),$e_name,$q_name,$k_route); 
$msg = array('test','123'); 
$re = $amqp->send($msg);

//get.php（接收并处理文件|消费者） 
 
require_once('amqp.php'); 
$config = require('config.php'); 
$config_qmqp = $config['amqp']; 
$e_name = 'e_guest'; //交换机名   
$k_route = 'k_route_sendemail'; //路由key  
$q_name = 'q_guest_sendemail'; //队列名 
$amqp = new Amqp($config_qmqp,$e_name,$q_name,$k_route); 
$re = $amqp->get();