TP5使用RabbitMQ實現(xiàn)消息隊列的項目實踐

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

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

1、安裝擴展

進入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()
? ? {
? ? ?? ?// 設置命令的名稱和描述
? ? ? ? $this->setName('test')->setDescription('這是一個測試命令');
? ? }
? ? /**
? ? ?* 執(zhí)行
? ? ?*/
? ? protected function execute(Input $input, Output $output)
? ? {
? ? ? ? $output->writeln("測試命令");
? ? }
}

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

第二步：

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

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

第三步：

測試命令，在項目根目錄下輸入命令：

php think test

回車運行之后輸出：

test command

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

3、rabbitmq服務端

下來我們自定義 RabbitMQ 啟動命令，守護進程運行，啟動 rabbirmq 服務端接收消息。
在 application/api/command 目錄下，新建 Ramq.php 文件，在執(zhí)行命令的方法中，調用 RabbitMQ 啟動守護進程方法即可。

<?php
namespace app\api\command;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use think\console\Command;
use think\console\Input;
use think\console\Output;
/**
?* RabbitMq 啟動命令
?*/
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("消息隊列開始");
? ? ? ? $this->start();
? ? ? ? // 指令輸出
? ? ? ? $output->writeln('消費隊列結束');
? ? }
? ? /**
? ? ?* 關閉
? ? ?*/
? ? function shutdown($channel, $connection)
? ? {
? ? ? ? $channel->close();
? ? ? ? $connection->close();
? ? }
? ? /**
? ? ?* 回調處理信息
? ? ?*/
? ? function process_message($message)
? ? {
? ? ? ? if ($message->body !== 'quit') {
? ? ? ? ? ? echo $message->body;
? ? ? ? }
? ? ? ? //手動應答
? ? ? ? $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']);
? ? ? ? }
? ? }
? ? /**
? ? ?* 啟動 守護進程運行
? ? ?*/
? ? 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ā)送消息的控制器，實現(xiàn)消息隊列，具體代碼如下：

<?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);
? ? }
}