phprabbitmq封装,封装php的RabbitMq

static private $_instance;static private $_conn;static private $amp;static private $route &＃61; ‘key_1‘;static private $q;static private $ex;static private $queue;public static functiongetInstance(){global $arr;if (!(self::$_instanceinstanceof self)) {

self::$_instance &＃61; new self($arr[‘RabbitMq‘]);return self::$_instance;

}return self::$_instance;

}private function __construct($conn)

{//创建连接和channel

$conn &＃61; new AMQPConnection($conn);if(!$conn->connect()) {die("Cannot connect to the broker!\n");

}

self::$_conn &＃61; new AMQPChannel($conn);

self::$amp &＃61; $conn;

}/**

*

* parm 交换机名

* parm 队列名

*

**/

public function listen($exchangeName,$queuename){

self::$queue &＃61; $queuename;return $this->setExchange($exchangeName,$queuename);

}//连接交换机

public function setExchange($exchangeName,$queueName){//创建交换机

$ex &＃61; new AMQPExchange(self::$_conn);

self::$ex &＃61; $ex;$ex->setName($exchangeName);$ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型

$ex->setFlags(AMQP_DURABLE); //持久化

$ex->declare();return self::setQueue($queueName,$exchangeName);

}//创建队列

private static function setQueue($queueName,$exchangeName){//创建队列

$q &＃61; new AMQPQueue(self::$_conn);$q->setName($queueName);$q->setFlags(AMQP_DURABLE);$q->declareQueue();//用于绑定队列和交换机

$routingKey &＃61; self::$route;$q->bind($exchangeName, $routingKey);

self::$q &＃61; $q;return(self::$_instance);

}/** 消费者

* $fun_name &＃61; array($classobj,$function) or function name string

* $autoack 是否自动应答

*

* function processMessage($envelope, $queue) {

$msg &＃61; $envelope->getBody();

echo $msg."\n"; //处理消息

$queue->ack($envelope->getDeliveryTag());//手动应答

}*/

public function run($func, $autoack &＃61; True){if (!$func || !self::$q) return False;while(True){if ($autoack) {if(!self::$q->consume($func,AMQP_AUTOACK)){//self::$q->ack($envelope->getDeliveryTag());

echo 123;

}

}

self::$q->consume($func);

}

}private static functioncloseConn(){

self::$amp->disconnect();

}public function pushlish($msg){while (1) {sleep(1);if (self::$ex->publish(date(‘H:i:s‘) . "用户" . "注册", self::$route)) {//写入文件等操作

echo $msg;

}

}

}//__clone方法防止对象被复制克隆

public function__clone()

{trigger_error(‘Clone is not allow!‘, E_USER_ERROR);

}

}