目錄1、安裝擴(kuò)展2、自定義命令3、rabbitmq服務(wù)端4、發(fā)送端5、驗(yàn)證

在使用 RabbitMQ 之前，你要安裝好 RabbitMQ 服務(wù)，具體安裝方法可以參考 windows下安裝RabbitMQ

1、安裝擴(kuò)展

進(jìn)入TP5 更目錄下，輸入命令安裝：

composer require php-amqplib/php-amqplib2、自定義命令

TP5 的自定義命令，這里也簡(jiǎn)單說(shuō)下。

第一步：

創(chuàng)建命令類(lèi)文件，新建 application/api/command/Test.php。

<?phpnamespace app\api\command;use think\console\Command;use think\console\Input;use think\console\Output;/**?* 自定義命令測(cè)試?*/class Test extends Command{?? ?/**?? ? * 配置?? ? */?? ?protected function configure()? ? {? ? ?? ?// 設(shè)置命令的名稱(chēng)和描述? ? ? ? $this->setName('test')->setDescription('這是一個(gè)測(cè)試命令');? ? }? ? /**? ? ?* 執(zhí)行? ? ?*/? ? protected function execute(Input $input, Output $output)? ? {? ? ? ? $output->writeln('測(cè)試命令');? ? }}

這個(gè)文件定義了一個(gè)叫test的命令，備注為 這是一個(gè)測(cè)試命令，執(zhí)行命令會(huì)輸出：test command。

第二步：

配置 command.php文件，在 application/command.php文件中添加命令。

<?phpreturn [?? ?'app\api\command\Test',];

第三步：

測(cè)試命令，在項(xiàng)目根目錄下輸入命令：

php think test

回車(chē)運(yùn)行之后輸出：

test command

到這里，自定義命令就結(jié)束了，test命令就自定義成功了。

3、rabbitmq服務(wù)端

下來(lái)我們自定義 RabbitMQ 啟動(dòng)命令，守護(hù)進(jìn)程運(yùn)行，啟動(dòng) rabbirmq 服務(wù)端接收消息。在 application/api/command 目錄下，新建 Ramq.php 文件，在執(zhí)行命令的方法中，調(diào)用 RabbitMQ 啟動(dòng)守護(hù)進(jìn)程方法即可。

<?phpnamespace app\api\command;use PhpAmqpLib\Connection\AMQPStreamConnection;use think\console\Command;use think\console\Input;use think\console\Output;/**?* RabbitMq 啟動(dòng)命令?*/class Ramq extends Command{?? ?protected $consumerTag = 'customer';? ? protected $exchange = 'xcuser';? ? protected $queue = 'xcmsg';?? ?protected function configure()? ? {? ? ? ? $this->setName('ramq')->setDescription('rabbitmq');? ? }? ? protected function execute(Input $input, Output $output)? ? {? ? ? ? $output->writeln('消息隊(duì)列開(kāi)始');? ? ? ? $this->start();? ? ? ? // 指令輸出? ? ? ? $output->writeln('消費(fèi)隊(duì)列結(jié)束');? ? }? ? /**? ? ?* 關(guān)閉? ? ?*/? ? function shutdown($channel, $connection)? ? {? ? ? ? $channel->close();? ? ? ? $connection->close();? ? }? ? /**? ? ?* 回調(diào)處理信息? ? ?*/? ? function process_message($message)? ? {? ? ? ? if ($message->body !== 'quit') {? ? ? ? ? ? echo $message->body;? ? ? ? }? ? ? ? //手動(dòng)應(yīng)答? ? ? ? $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);? ? ? ? if ($message->body === 'quit') {? ? ? ? ? ? $message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']);? ? ? ? }? ? }? ? /**? ? ?* 啟動(dòng) 守護(hù)進(jìn)程運(yùn)行? ? ?*/? ? public function start()? ? {? ? ? ? $host = '127.0.0.1';? ? ? ? $port = 5672;? ? ? ? $user = 'guest';? ? ? ? $pwd = 'guest';? ? ? ? $vhost = '/';? ? ? ? $connection = new AMQPStreamConnection($host, $port, $user, $pwd, $vhost);? ? ? ? $channel = $connection->channel();? ? ? ? $channel->queue_declare($this->queue, false, true, false, false);? ? ? ? $channel->exchange_declare($this->exchange, 'direct', false, true, false);? ? ? ? $channel->queue_bind($this->queue, $this->exchange);? ? ? ? $channel->basic_consume($this->queue, $this->consumerTag, false, false, false, false, array($this, 'process_message'));? ? ? ? register_shutdown_function(array($this, 'shutdown'), $channel, $connection);? ? ? ? while (count($channel->callbacks)) {? ? ? ? ? ? $channel->wait();? ? ? ? }? ? }}

在application/command.php文件中，添加rabbitmq自定義命令。

return [ 'app\api\command\Ramq',// rabbitmq];4、發(fā)送端

最后，我們?cè)賹?xiě)發(fā)送消息的控制器，實(shí)現(xiàn)消息隊(duì)列，具體代碼如下：

<?phpnamespace app\api\controller;use PhpAmqpLib\Connection\AMQPStreamConnection;use PhpAmqpLib\Message\AMQPMessage;use think\Controller;/**?* 發(fā)送端?*/class MessageQueue extends Controller{?? ?const exchange = 'xcuser';? ? const queue = 'xcmsg';? ? /**? ? ?* 發(fā)送消息? ? ?*/? ? public function pushMessage($data)? ? {? ? ? ? $host = '127.0.0.1';? ? ? ? $port = 5672;? ? ? ? $user = 'guest';? ? ? ? $pwd = 'guest';? ? ? ? $vhost = '/';? ? ? ? $connection = new AMQPStreamConnection($host, $port, $user, $pwd, $vhost);? ? ? ? $channel = $connection->channel();? ? ? ? $channel->exchange_declare(self::exchange, 'direct', false, true, false);? ? ? ? $channel->queue_declare(self::queue, false, true, false, false);? ? ? ? $channel->queue_bind(self::queue, self::exchange);? ? ? ? $messageBody = $data;? ? ? ? $message = new AMQPMessage($messageBody, array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));? ? ? ? $channel->basic_publish($message, self::exchange);? ? ? ? $channel->close();? ? ? ? $connection->close();? ? ? ? echo 'ok';? ? }? ? /**? ? ?* 執(zhí)行? ? ?*/? ? public function index()? ? {? ? ? ? $data = json_encode(['msg' => '測(cè)試數(shù)據(jù)', 'id' => '15']);? ? ? ? $this->pushMessage($data);? ? }}5、驗(yàn)證

先執(zhí)行自定義命令，啟動(dòng) rabbitmq 守護(hù)進(jìn)程。在項(xiàng)目更目錄下打開(kāi)命令行，輸入下面命令：

php think ramq

然后在瀏覽器訪(fǎng)問(wèn)發(fā)送信息的方法，http://你的域名/api/message/index，你發(fā)送一次消息，在命令行就會(huì)輸出一條消息。這樣我們就用 RabbitMQ 實(shí)現(xiàn)了一個(gè)簡(jiǎn)單的消息隊(duì)列。

到此這篇關(guān)于TP5使用RabbitMQ實(shí)現(xiàn)消息隊(duì)列的項(xiàng)目實(shí)踐的文章就介紹到這了,更多相關(guān)TP5 RabbitMQ消息隊(duì)列內(nèi)容請(qǐng)搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！