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."\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生产者代码





<?php

set_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消费者代码





<?php

error_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.''."\r\n",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程序设计有所帮助。