您好,登錄后才能下訂單哦!
怎么在PHP中使用RabbitMQ實(shí)現(xiàn)消息隊(duì)列?針對(duì)這個(gè)問(wèn)題,這篇文章詳細(xì)介紹了相對(duì)應(yīng)的分析和解答,希望可以幫助更多想解決這個(gè)問(wèn)題的小伙伴找到更簡(jiǎn)單易行的方法。
config.php
<?php return [ //配置 'host' => [ 'host' => '127.0.0.1', 'port' => '5672', 'login' => 'guest', 'password' => 'guest', 'vhost'=>'/', ], //交換機(jī) 'exchange'=>'word', //路由 'routes' => [], ];
BaseMQ.php
<?php /** * Created by PhpStorm. * User: pc * Date: 2018/12/13 * Time: 14:11 */ namespace MyObjSummary\rabbitMQ; /** Member * AMQPChannel * AMQPConnection * AMQPEnvelope * AMQPExchange * AMQPQueue * Class BaseMQ * @package MyObjSummary\rabbitMQ */ class BaseMQ { /** MQ Channel * @var \AMQPChannel */ public $AMQPChannel ; /** MQ Link * @var \AMQPConnection */ public $AMQPConnection ; /** MQ Envelope * @var \AMQPEnvelope */ public $AMQPEnvelope ; /** MQ Exchange * @var \AMQPExchange */ public $AMQPExchange ; /** MQ Queue * @var \AMQPQueue */ public $AMQPQueue ; /** conf * @var */ public $conf ; /** exchange * @var */ public $exchange ; /** link * BaseMQ constructor. * @throws \AMQPConnectionException */ public function __construct() { $conf = require 'config.php' ; if(!$conf) throw new \AMQPConnectionException('config error!'); $this->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!\n"); } /** * 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
<?php //生產(chǎn)者 P namespace MyObjSummary\rabbitMQ; require 'BaseMQ.php'; class ProductMQ extends BaseMQ { private $routes = ['hello','word']; //路由key /** * ProductMQ constructor. * @throws \AMQPConnectionException */ public function __construct() { parent::__construct(); } /** 只控制發(fā)送成功 不接受消費(fèi)者是否收到 * @throws \AMQPChannelException * @throws \AMQPConnectionException * @throws \AMQPExchangeException */ public function run() { //頻道 $channel = $this->channel(); //創(chuàng)建交換機(jī)對(duì)象 $ex = $this->exchange(); //消息內(nèi)容 $message = 'product message '.rand(1,99999); //開(kāi)始事務(wù) $channel->startTransaction(); $sendEd = true ; foreach ($this->routes as $route) { $sendEd = $ex->publish($message, $route) ; echo "Send Message:".$sendEd."\n"; } if(!$sendEd) { $channel->rollbackTransaction(); } $channel->commitTransaction(); //提交事務(wù) $this->close(); die ; } } try{ (new ProductMQ())->run(); }catch (\Exception $exception){ var_dump($exception->getMessage()) ; }
ConsumerMQ.php
<?php //消費(fèi)者 C namespace MyObjSummary\rabbitMQ; require 'BaseMQ.php'; class ConsumerMQ extends BaseMQ { private $q_name = 'hello'; //隊(duì)列名 private $route = 'hello'; //路由key /** * ConsumerMQ constructor. * @throws \AMQPConnectionException */ public function __construct() { parent::__construct(); } /** 接受消息 如果終止 重連時(shí)會(huì)有消息 * @throws \AMQPChannelException * @throws \AMQPConnectionException * @throws \AMQPExchangeException * @throws \AMQPQueueException */ public function run() { //創(chuàng)建交換機(jī) $ex = $this->exchange(); $ex->setType(AMQP_EX_TYPE_DIRECT); //direct類型 $ex->setFlags(AMQP_DURABLE); //持久化 //echo "Exchange Status:".$ex->declare()."\n"; //創(chuàng)建隊(duì)列 $q = $this->queue(); //var_dump($q->declare());exit(); $q->setName($this->q_name); $q->setFlags(AMQP_DURABLE); //持久化 //echo "Message Total:".$q->declareQueue()."\n"; //綁定交換機(jī)與隊(duì)列,并指定路由鍵 echo 'Queue Bind: '.$q->bind($this->exchange, $this->route)."\n"; //阻塞模式接收消息 echo "Message:\n"; while(True){ $q->consume(function ($envelope,$queue){ $msg = $envelope->getBody(); echo $msg."\n"; //處理消息 $queue->ack($envelope->getDeliveryTag()); //手動(dòng)發(fā)送ACK應(yīng)答 }); //$q->consume('processMessage', AMQP_AUTOACK); //自動(dòng)ACK應(yīng)答 } $this->close(); } } try{ (new ConsumerMQ)->run(); }catch (\Exception $exception){ var_dump($exception->getMessage()) ; }
php的框架:1、Laravel,Laravel是一款免費(fèi)并且開(kāi)源的PHP應(yīng)用框架。2、Phalcon,Phalcon是運(yùn)行速度最快的一個(gè)PHP框架。3、Symfony,Symfony是一款為Web項(xiàng)目準(zhǔn)備的PHP框架。4、Yii,Yii是一款快速、安全和專業(yè)的PHP框架。5、CodeIgniter,CodeIgniter是一款非常敏捷的開(kāi)源PHP框架。6、CakePHP,CakePHP是一款老牌的PHP框架。7.Kohana,Kohana是一款敏捷但是功能強(qiáng)大的PHP框架。
關(guān)于怎么在PHP中使用RabbitMQ實(shí)現(xiàn)消息隊(duì)列問(wèn)題的解答就分享到這里了,希望以上內(nèi)容可以對(duì)大家有一定的幫助,如果你還有很多疑惑沒(méi)有解開(kāi),可以關(guān)注億速云行業(yè)資訊頻道了解更多相關(guān)知識(shí)。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。