Amqp.class.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;

        //創(chuàng)建連接和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();
    }

    //創(chuàng)建交換機(jī)
    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";   //隊列內(nèi)容總數(shù)
        $ex->declare();
        $this->ex = $ex;
    }
    
    //創(chuàng)建隊列   
    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";  
        //綁定交換機(jī)與隊列，并指定路由鍵  
        //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);
    }
    
    //發(fā)送消息
    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發(fā)送
        $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(加入隊列文件|生產(chǎn)者)

require_once('Amqp.class.php');
$e_name  = 'e_guest';            //交換機(jī)名  
$k_route = 'k_route_feedpush';   //路由key
$q_name  = 'q_guest_feedpush';   //隊列名
$config  = config('amqp');       //導(dǎo)入配置文件

$amqp = new Amqp(config('amqp'),$e_name,$q_name,$k_route);

$msg = array('test','123');

$re = $amqp->send($msg);

get.php（接收并處理文件|消費者）

require_once('Amqp.class.php');
$config = require('config.php');
$config_qmqp = $config['amqp'];

$e_name  = 'e_guest'; //交換機(jī)名  
$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();