PHP+RabbitMQメッセージキューの完全なコードを実現

なぜActiveMqやRocketMqではなくRabbitMqを使うのですか?まず、業務上、私はメッセージの100%の受け入れ率を要求していません.そして、php開発と結びつけて、RabbitMqはRocketMqに比べて、遅延が低い(微妙なレベル)必要があります.ActiveMqについては、問題が多いようです.RabbitMqは様々な言語のサポートが良いので、RabbitMqを選びます.まずPHP対応のRabbitMQを取り付け、ここではphp_amqpの異なる拡張実現方式には微細な違いがある.

php    : http://pecl.php.net/package/amqp
         http://www.rabbitmq.com/getstarted.html 

紹介するphp構成情報
BaseMQ.php MQベースクラス
ProductMQ.php生産者クラス
ConsumerMQ.php消費者クラス
Consumer2MQ.php消費者2(複数可)
config.php

 [
  'host' => '127.0.0.1',
  'port' => '5672',
  'login' => 'guest',
  'password' => 'guest',
  'vhost'=>'/',
 ],
 //   
 'exchange'=>'word',
 //  
 'routes' => [],
];

BaseMQ.php

conf  = $conf['host'] ;
  $this->exchange = $conf['exchange'] ;
  $this->AMQPConnection = new \AMQPConnection($this->conf);
  if (!$this->AMQPConnection->connect())
   throw new \AMQPConnectionException("Cannot connect to the broker!
");
 }

 /**
  * close link
  */
 public function close()
 {
  $this->AMQPConnection->disconnect();
 }

 /** Channel
  * @return \AMQPChannel
  * @throws \AMQPConnectionException
  */
 public function channel()
 {
  if(!$this->AMQPChannel) {
   $this->AMQPChannel = new \AMQPChannel($this->AMQPConnection);
  }
  return $this->AMQPChannel;
 }

 /** Exchange
  * @return \AMQPExchange
  * @throws \AMQPConnectionException
  * @throws \AMQPExchangeException
  */
 public function exchange()
 {
  if(!$this->AMQPExchange) {
   $this->AMQPExchange = new \AMQPExchange($this->channel());
   $this->AMQPExchange->setName($this->exchange);
  }
  return $this->AMQPExchange ;
 }

 /** queue
  * @return \AMQPQueue
  * @throws \AMQPConnectionException
  * @throws \AMQPQueueException
  */
 public function queue()
 {
  if(!$this->AMQPQueue) {
   $this->AMQPQueue = new \AMQPQueue($this->channel());
  }
  return $this->AMQPQueue ;
 }

 /** Envelope
  * @return \AMQPEnvelope
  */
 public function envelope()
 {
  if(!$this->AMQPEnvelope) {
   $this->AMQPEnvelope = new \AMQPEnvelope();
  }
  return $this->AMQPEnvelope;
 }
}

ProductMQ.php

channel();
  //       
  $ex = $this->exchange();
  //    
  $message = 'product message '.rand(1,99999);
  //    
  $channel->startTransaction();
  $sendEd = true ;
  foreach ($this->routes as $route) {
   $sendEd = $ex->publish($message, $route) ;
   echo "Send Message:".$sendEd."
";
  }
  if(!$sendEd) {
   $channel->rollbackTransaction();
  }
  $channel->commitTransaction(); //    
  $this->close();
  die ;
 }
}
try{
 (new ProductMQ())->run();
}catch (\Exception $exception){
 var_dump($exception->getMessage()) ;
}

ConsumerMQ.php

exchange();
  $ex->setType(AMQP_EX_TYPE_DIRECT); //direct  
  $ex->setFlags(AMQP_DURABLE); //   
  //echo "Exchange Status:".$ex->declare()."
";

  //    
  $q = $this->queue();
  //var_dump($q->declare());exit();
  $q->setName($this->q_name);
  $q->setFlags(AMQP_DURABLE); //   
  //echo "Message Total:".$q->declareQueue()."
";

  //        ，      
  echo 'Queue Bind: '.$q->bind($this->exchange, $this->route)."
";

  //        
  echo "Message:
";
  while(True){
   $q->consume(function ($envelope,$queue){
    $msg = $envelope->getBody();
    echo $msg."
"; //    
    $queue->ack($envelope->getDeliveryTag()); //    ACK  
   });
   //$q->consume('processMessage', AMQP_AUTOACK); //  ACK  
  }
  $this->close();
 }
}
try{
 (new ConsumerMQ)->run();
}catch (\Exception $exception){
 var_dump($exception->getMessage()) ;
}

phperは進級する時いつもいくつかの問題とボトルネックに出会って、業務コードが書きすぎて方向感がなくて、給料を上げる必要がありますが、そこから昇進すべきか分かりません.これに対して私はいくつかの資料を整理しました.分布式アーキテクチャ、高拡張性、高性能、高同時性、サーバー性能の調整、TP 6、laravel、YII 2、Redis、Swoole、Kafka、Mysqlの最適化、shellスクリプト、Docker、マイクロサービス、Nginxなど複数の知識ポイント高級進級干物に必要なものは無料で共有できますので、こちらをスタンプしてください