RabbitMQ学习之（五）_一个基于PHP的RabbitMQ操作类详解程序员

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()."/n"; //队列内容总数 $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()."/n"; //绑定交换机与队列，并指定路由键 //echo "queue status: ".$q- declare(); //echo "/n"; //echo Queue Bind: .$q- bind($this- e_name, $this- k_route)."/n"; //echo "/n"; $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 "==========/n"; $messages = $q- get(AMQP_AUTOACK); $return[]=json_decode($messages- getBody(),true); //echo "/n"; $this- conn- disconnect(); return $return; }