TestMqWorker.php 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135
  1. <?php
  2. namespace App\Console\Commands;
  3. use App\Tools\Tools;
  4. use Illuminate\Console\Command;
  5. use PhpAmqpLib\Channel\AMQPChannel;
  6. use PhpAmqpLib\Connection\AbstractConnection;
  7. use PhpAmqpLib\Connection\AMQPStreamConnection;
  8. use PhpAmqpLib\Exchange\AMQPExchangeType;
  9. use PhpAmqpLib\Message\AMQPMessage;
  10. class TestMqWorker extends Command
  11. {
  12. /**
  13. * The name and signature of the console command.
  14. * php artisan test:mq.worker
  15. *
  16. * @var string
  17. */
  18. protected $signature = 'test:mq.worker';
  19. /**
  20. * The console command description.
  21. *
  22. * @var string
  23. */
  24. protected $description = 'Command description';
  25. /**
  26. * Create a new command instance.
  27. *
  28. * @return void
  29. */
  30. public function __construct()
  31. {
  32. parent::__construct();
  33. }
  34. /**
  35. * Execute the console command.
  36. *
  37. * @return int
  38. */
  39. public function handle()
  40. {
  41. if (Tools::isStop()) {
  42. return 0;
  43. }
  44. $exchange = 'router';
  45. $queue = 'hello';
  46. $consumerTag = 'consumer';
  47. $connection = new AMQPStreamConnection(config('queue.connections.rabbitmq.host'),
  48. config('queue.connections.rabbitmq.port'),
  49. config('queue.connections.rabbitmq.user'),
  50. config('queue.connections.rabbitmq.password'),
  51. config('queue.connections.rabbitmq.virtual_host'));
  52. $channel = $connection->channel();
  53. /*
  54. The following code is the same both in the consumer and the producer.
  55. In this way we are sure we always have a queue to consume from and an
  56. exchange where to publish messages.
  57. */
  58. /*
  59. name: $queue
  60. passive: false
  61. durable: true // the queue will survive server restarts
  62. exclusive: false // the queue can be accessed in other channels
  63. auto_delete: false //the queue won't be deleted once the channel is closed.
  64. */
  65. $channel->queue_declare($queue, false, true, false, false);
  66. /*
  67. name: $exchange
  68. type: direct
  69. passive: false
  70. durable: true // the exchange will survive server restarts
  71. auto_delete: false //the exchange won't be deleted once the channel is closed.
  72. */
  73. $channel->exchange_declare($exchange, AMQPExchangeType::DIRECT, false, true, false);
  74. $channel->queue_bind($queue, $exchange);
  75. /**
  76. * @param AMQPMessage $message
  77. */
  78. $process_message = function ($message) {
  79. echo "\n--------\n";
  80. echo $message->body;
  81. echo "\n--------\n";
  82. $message->ack();
  83. // Send a message with the string "quit" to cancel the consumer.
  84. if ($message->body === 'quit') {
  85. $message->getChannel()->basic_cancel($message->getConsumerTag());
  86. }
  87. };
  88. /*
  89. queue: Queue from where to get the messages
  90. consumer_tag: Consumer identifier
  91. no_local: Don't receive messages published by this consumer.
  92. no_ack: If set to true, automatic acknowledgement mode will be used by this consumer. See https://www.rabbitmq.com/confirms.html for details.
  93. exclusive: Request exclusive consumer access, meaning only this consumer can access the queue
  94. nowait:
  95. callback: A PHP Callback
  96. */
  97. $channel->basic_consume($queue, $consumerTag, false, false, false, false, $process_message);
  98. /**
  99. * @param AMQPChannel $channel
  100. * @param AbstractConnection $connection
  101. */
  102. $shutdown = function ($channel, $connection) {
  103. $channel->close();
  104. $connection->close();
  105. };
  106. register_shutdown_function($shutdown, $channel, $connection);
  107. // Loop as long as the channel has callbacks registered
  108. while ($channel->is_consuming()) {
  109. $channel->wait(null, true);
  110. // do something else
  111. usleep(300000);
  112. }
  113. return 0;
  114. }
  115. }