PHP基于rabbitmq操作类的生产者和消费者功能示例

本文实例讲述了PHP基于rabbitmq操作类的生产者和消费者功能。分享给大家供大家参考，具体如下：

注意事项：

1、accept.php消费者代码需要在命令行执行

2、"username"=>"asdf","password"=>"123456" 改成自己的帐号和密码

RabbitMQCommand.php操作类代码

<?php/* * amqp协议操作类，可以访问rabbitMQ * 需先安装php_amqp扩展 */class RabbitMQCommand{  public $configs = array();  //交换机名称  public $exchange_name = "";  //队列名称  public $queue_name = "";  //路由名称  public $route_key = "";  /*   * 持久化，默认True   */  public $durable = True;  /*   * 自动删除   * exchange is deleted when all queues have finished using it   * queue is deleted when last consumer unsubscribes   *   */  public $autodelete = False;  /*   * 镜像   * 镜像队列，打开后消息会在节点之间复制，有master和slave的概念   */  public $mirror = False;  private $_conn = Null;  private $_exchange = Null;  private $_channel = Null;  private $_queue = Null;  /*   * @configs array("host"=>$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.""; //处理消息      $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生产者代码

<?phpset_time_limit(0);include_once("RabbitMQCommand.php");$configs = array("host"=>"127.0.0.1","port"=>5672,"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,$queue_name,$route_key);for($i=0;$i<=100;$i++){  $ra->send(date("Y-m-d H:i:s",time()));}exit();

accept.php消费者代码

<?phperror_reporting(0);include_once("RabbitMQCommand.php");$configs = array("host"=>"127.0.0.1","port"=>5672,"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,$queue_name,$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.""."",FILE_APPEND);    $queue->ack($envelopeID);  }}$a = new A();$s = $ra->run(array($a,"processMessage"),false);

更多关于PHP相关内容感兴趣的读者可查看本站专题：《PHP数据结构与算法教程》、《php程序设计算法总结》、《php字符串(string)用法总结》、《PHP数组(Array)操作技巧大全》、《PHP常用遍历算法与技巧总结》及《PHP数学运算技巧总结》

希望本文所述对大家PHP程序设计有所帮助。