php消息队列处理实践 ，利用AMQP和redis两种方法

类 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' => 'localhost',
'port' => '5672',
'vhost' => '/',
'user' => 'guest',
'password' => 'guest'
)
),
);

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

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

//redis出队列POP
function actionRedisPop()
{
$redis = new Redis;

$redis->connect('cloud_redis',9002);

while ($usr = $redis->rPop('list_test')) {
$array = json_decode($usr,true);
print_R($array);
}

}

//redis入队列push
function actionRedisPush()
{
$redis = new Redis;

$redis->connect('cloud_redis',9002);

$data = array('list_name'=>'usr','value'=>date('Y-m-d H:i:s'));
$json = json_encode($data);
var_dump($redis->lPush('list_test', $json));

}