TP5使用RabbitMQ實(shí)現(xiàn)消息隊(duì)列的項(xiàng)目實(shí)踐

更新時(shí)間：2023年06月29日 09:41:17 作者：L小臣

本文主要介紹了TP5使用RabbitMQ實(shí)現(xiàn)消息隊(duì)列的項(xiàng)目實(shí)踐，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

1、安裝擴(kuò)展

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

composer require php-amqplib/php-amqplib

2、自定義命令

TP5 的自定義命令，這里也簡單說下。

第一步：

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

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

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

第二步：

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

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

第三步：

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

php think test

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

test command

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

3、rabbitmq服務(wù)端

下來我們自定義 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)程方法即可。

<?php
namespace 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ì)列開始");
? ? ? ? $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ā)送端

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

<?php
namespace 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' => '測試數(shù)據(jù)', 'id' => '15']);
? ? ? ? $this->pushMessage($data);
? ? }
}

5、驗(yàn)證

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

php think ramq

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

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

您可能感興趣的文章: