1
0

Message.php 2.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120
  1. <?php
  2. namespace common\logic\Amqp;
  3. use common\helpers\KeyHelper;
  4. use PhpAmqpLib\Message\AMQPMessage;
  5. use Yii;
  6. /**
  7. * 处理消息体逻辑
  8. * Class Message
  9. * @package common\logic\MQMessage
  10. * @author libingke
  11. */
  12. class Message extends Connect
  13. {
  14. private $_connect;
  15. private $_channel;
  16. private $_callback_queue;
  17. private $_response;
  18. private $_corr_id;
  19. public function __construct($queueName = '')
  20. {
  21. $this->_connect = self::connect();
  22. $this->_channel = $this->_connect->channel();
  23. $this->_callback_queue = $queueName;
  24. }
  25. /**
  26. * [发送信息]
  27. * @author: libingke
  28. * @param string $body 消息主体
  29. * @param string $routing_key
  30. * @return bool
  31. */
  32. public function send($body, $routing_key)
  33. {
  34. $this->_corr_id = KeyHelper::getUniqueId('message_send');
  35. $properties = [
  36. //'content_type' => 'text/plain',
  37. 'correlation_id' => $this->_corr_id,
  38. 'reply_to' => $this->_callback_queue
  39. ];
  40. if (is_string($body)) {
  41. $msg = new AMQPMessage( (string) $body, $properties);
  42. $this->_channel->basic_publish($msg, '', $routing_key);
  43. }
  44. return $this->_corr_id;
  45. }
  46. /**
  47. * [批量发送信息]
  48. * @author: libingke
  49. * @param $info
  50. * @param $queue
  51. */
  52. public function batchSend($info, $queue)
  53. {
  54. foreach ($info as $str) {
  55. $this->_channel->batch_basic_publish($str, '', $queue);
  56. }
  57. $this->_channel->publish_batch();
  58. }
  59. /**
  60. * [消费信息] todo
  61. * @author: libingke
  62. * @param string $queue 队列名称
  63. * @param string $str
  64. * @return bool
  65. */
  66. public function consume($queue)
  67. {
  68. $callback = function($msg) {
  69. echo " [x] Received ", $msg->body, "\n";
  70. $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
  71. };
  72. $this->_channel->basic_qos(null, 1, null);
  73. $this->_channel->basic_consume(
  74. $queue, '', false, false, false, false,
  75. $callback
  76. );
  77. while(count($this->_channel->callbacks)) {
  78. $this->_channel->wait();
  79. }
  80. return true;
  81. }
  82. /**
  83. * [批量消费信息]
  84. * @author: libingke
  85. * @param string $queue 队列名称
  86. * @param string $arr
  87. * @return array 返回数据
  88. */
  89. public function batchConsume($queue, $arr)
  90. {
  91. }
  92. /**
  93. * [删除信息]
  94. * @author: libingke
  95. */
  96. public function delete($message)
  97. {
  98. }
  99. }