这篇文章主要介绍了PHP基于rabbitmq操作类的生产者和消费者功能,结合实例形式分析了基于rabbitmq操作类的生产者和消费者定义与使用方法,需要的朋友可以参考下
本文实例讲述了PHP基于rabbitmq操作类的生产者和消费者功能。分享给大家供大家参考，具体如下：
注意事项：
1、accept.PHP消费者代码需要在命令行执行
2、'username'=>'asdf','password'=>'123456' 改成自己的帐号和密码
RabbitMQCommand.PHP操作类代码
$host,'port'=>5672,'username'=>$username,'password'=>$password,'vhost'=>'/')
*/
public function __construct($configs = array(),$exchange_name = '',$queue_name = '',$route_key = '') {
$this->setConfigs($configs);
$this->exchange_name = $exchange_name;
$this->queue_name = $queue_name;
$this->route_key = $route_key;
}
private function setConfigs($configs) {
if (!is_array($configs)) {
throw new Exception('configs is not array');
}
if (!($configs['host'] && $configs['port'] && $configs['username'] && $configs['password'])) {
throw new Exception('configs is empty');
}
if (empty($configs['vhost'])) {
$configs['vhost'] = '/';
}
$configs['login'] = $configs['username'];
unset($configs['username']);
$this->configs = $configs;
}
/*
* 设置是否持久化，默认为True
*/
public function setDurable($durable) {
$this->durable = $durable;
}
/*
* 设置是否自动删除
*/
public function setAutoDelete($autodelete) {
$this->autodelete = $autodelete;
}
/*
* 设置是否镜像
*/
public function setMirror($mirror) {
$this->mirror = $mirror;
}
/*
* 打开amqp连接
*/
private function open() {
if (!$this->_conn) {
try {
$this->_conn = new AMQPConnection($this->configs);
$this->_conn->connect();
$this->initConnection();
} catch (AMQPConnectionException $ex) {
throw new Exception('cannot connection rabbitmq',500);
}
}
}
/*
* rabbitmq连接不变
* 重置交换机，队列，路由等配置
*/
public function reset($exchange_name,$queue_name,$route_key) {
$this->exchange_name = $exchange_name;
$this->queue_name = $queue_name;
$this->route_key = $route_key;
$this->initConnection();
}
/*
* 初始化rabbit连接的相关配置
*/
private function initConnection() {
if (empty($this->exchange_name) || empty($this->queue_name) || empty($this->route_key)) {
throw new Exception('rabbitmq exchange_name or queue_name or route_key is empty',500);
}
$this->_channel = new AMQPChannel($this->_conn);
$this->_exchange = new AMQPExchange($this->_channel);
$this->_exchange->setName($this->exchange_name);
$this->_exchange->setType(AMQP_EX_TYPE_DIRECT);
if ($this->durable)
$this->_exchange->setFlags(AMQP_DURABLE);
if ($this->autodelete)
$this->_exchange->setFlags(AMQP_AUTODELETE);
$this->_exchange->declare();
$this->_queue = new AMQPQueue($this->_channel);
$this->_queue->setName($this->queue_name);
if ($this->durable)
$this->_queue->setFlags(AMQP_DURABLE);
if ($this->autodelete)
$this->_queue->setFlags(AMQP_AUTODELETE);
if ($this->mirror)
$this->_queue->setArgument('x-ha-policy','all');
$this->_queue->declare();
$this->_queue->bind($this->exchange_name,$this->route_key);
}
public function close() {
if ($this->_conn) {
$this->_conn->disconnect();
}
}
public function __sleep() {
$this->close();
return array_keys(get_object_vars($this));
}
public function __destruct() {
$this->close();
}
/*
* 生产者发送消息
*/
public function send($msg) {
$this->open();
if(is_array($msg)){
$msg = json_encode($msg);
}else{
$msg = trim(strval($msg));
}
return $this->_exchange->publish($msg,$this->route_key);
}
/*
* 消费者
* $fun_name = array($classobj,$function) or function name string
* $autoack 是否自动应答
*
* function processMessage($envelope,$queue) {
$msg = $envelope->getBody();
echo $msg."\n"; //处理消息
$queue->ack($envelope->getDeliveryTag());//手动应答
}
*/
public function run($fun_name,$autoack = True){
$this->open();
if (!$fun_name || !$this->_queue) return False;
while(True){
if ($autoack) $this->_queue->consume($fun_name,AMQP_AUTOACK);
else $this->_queue->consume($fun_name);
}
}
}
send.PHP生产者代码
'127.0.0.1','username'=>'asdf','password'=>'123456','vhost'=>'/');
$exchange_name = 'class-e-1';
$queue_name = 'class-q-1';
$route_key = 'class-r-1';
$ra = new RabbitMQCommand($configs,$exchange_name,$route_key);
for($i=0;$i<=100;$i++){
$ra->send(date('Y-m-d H:i:s',time()));
}
exit();
accept.PHP消费者代码
'127.0.0.1',$route_key);
class A{
function processMessage($envelope,$queue) {
$msg = $envelope->getBody();
$envelopeID = $envelope->getDeliveryTag();
$pid = posix_getpid();
file_put_contents("log{$pid}.log",$msg.'|'.$envelopeID.''."\r\n",FILE_APPEND);
$queue->ack($envelopeID);
}
}
$a = new A();
$s = $ra->run(array($a,'processMessage'),false);
更多关于PHP相关内容感兴趣的读者可查看本站专题：《》、《》、《》、《》、《》及《》
希望本文所述对大家PHP程序设计有所帮助。
服务热线 19970861797
