12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697 |
- <?php
- namespace App\Console\Commands;
- use App\Repositories\FeedRepositories;
- use Illuminate\Console\Command;
- use Illuminate\Support\Facades\Log;
- use PhpAmqpLib\Connection\AMQPStreamConnection;
- class ContentFeedCreate extends Command
- {
- /**
- * The name and signature of the console command.
- *
- * @var string
- */
- protected $signature = 'content:feed';
- /**
- * The console command description.
- *
- * @var string
- */
- protected $description = 'virus通知异步处理-feed生成,帖子数据统计';
- protected $connection;
- protected $channel;
- protected $queue = 'content_feed_queue';
- /**
- * Create a new command instance.
- *
- * @return void
- */
- public function __construct(FeedRepositories $feedRepositories)
- {
- parent::__construct();
- $this->feedRepositories = $feedRepositories;
- }
- public function getConnection()
- {
- $conn = false;
- if ($this->connection) {
- return $this->connection;
- }
- for ($i = 0; $i < 3; $i++) {
- $connection = $this->createConn();
- if ($connection) {
- $conn = $connection;
- break;
- }
- Log::info("create amqp conn retry=" . $i);
- }
- return $conn;
- }
- public function createConn()
- {
- try {
- $connection = new AMQPStreamConnection(env('MQ_HOST'), env('MQ_PORT'), env('MQ_USERNAME'), env('MQ_PWD'), env('MQ_VHOST'));
- } catch (\Exception $exception) {
- Log::info("AMQP connection Error" . $exception->getMessage());
- $connection = false;
- }
- return $connection;
- }
- /**
- * Execute the console command.
- *
- * @return mixed
- */
- public function handle()
- {
- $this->connection = $this->getConnection();
- $this->channel = $this->connection->channel();
- $this->channel->queue_declare($this->queue, false, true, false, false);
- $callback = function ($msg) {
- $param = \GuzzleHttp\json_decode($msg->body,true);
- $this->line(date('Y-m-d H:i:s').'收到消息:'.$msg->body);
- $this->feedRepositories->contentCreate($param);
- $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
- };
- $this->channel->basic_qos(null, 1, null);
- $this->channel->basic_consume($this->queue, '', false, false, false, false, $callback);
- while (count($this->channel->callbacks)) {
- $this->channel->wait();
- }
- $this->channel->close();
- $this->connection->close();
- }
- }
|