phpメッセージキュー処理の実践,AMQPとredisの2つの方法を利用する
一:AMQP方法を利用する
プロファイル:
ファイルの受信と処理:
amqp.php
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()."
"; //
$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()."
";
// ,
//echo "queue status: ".$q->declare();
//echo "
";
//echo 'Queue Bind: '.$q->bind($this->e_name, $this->k_route)."
";
//echo "
";
$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 "==========
";
$messages = $q->get(AMQP_AUTOACK);
$return[]=json_decode($messages->getBody(),true);
//echo "
";
}
$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を利用してメッセージキュー処理を行う//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)); }