1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677 |
- <?php
- namespace components;
- use PhpAmqpLib\Message\AMQPMessage;
- use PhpAmqpLib\Connection\AMQPStreamConnection;
- class PhpClient {
- private $connection;
- private $channel;
- private $callback_queue;
- private $response;
- private $corr_id;
- private $result = '';
- CONST HOST = "172.30.118.225";
- CONST PORT = 5673; //默认5672
- CONST USER = "guest"; //用户名
- CONST PASS = "guest";//密码
- public function __construct() {
- $this->connection = new AMQPStreamConnection(
- self::HOST, self::PORT, self::USER, self::PASS); //建立连接
- $this->channel = $this->connection->channel();
- list($this->callback_queue, ,) = $this->channel->queue_declare(
- "", false, false, true, false);
- $this->channel->basic_consume(
- $this->callback_queue, '', false, false, false, false,
- array($this, 'on_response'));
- }
- public function on_response($rep) {
- if($rep->get('correlation_id') == $this->corr_id) {
- $this->result .= $rep->body;
- $this->response = $rep->body;
- }
- }
- public function call($n) {
- $this->response = null;
- $this->corr_id = uniqid();
- $msg = new AMQPMessage(
- (string) $n,
- array('correlation_id' => $this->corr_id,
- 'reply_to' => $this->callback_queue)
- );
- $this->channel->basic_publish($msg, '', 'queue'); //这里的queue是消息名称
- while($this->response != "end") {
- $this->channel->wait();
- }
- return $this->result;
- }
- public static function CallMq($n){
- $connection = new AMQPStreamConnection(
- self::HOST, self::PORT, self::USER, self::PASS); //建立连接
- $channel = $connection->channel();
- $channel->queue_declare('task_queue', false, true, false, false);
- $data=empty($n)?"Hello World!":$n;
- $msg = new AMQPMessage($data,
- array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT)
- );
- $channel->basic_publish($msg, '', 'task_queue');
- $channel->close();
- $connection->close();
- return true;
- }
- }
|