03.tp6簡單應用 RabbitMQ · 初心（學習筆記）

>[info] tp6簡單應用 RabbitMQ **1. composer 安裝 AMQP 擴展** ``` composer require php-amqplib/php-amqplib ``` **2. 在 config 目錄下創建 rabbitmq.php 文件** ~~~ // rabbitmq 配置信息 return [ 'host'=>'127.0.0.1', 'port'=>'5672', 'user'=>'test', 'password'=>'123456', 'vhost'=>'/', 'exchange_name' => 'hello', 'queue_name' => 'hello', 'route_key' => 'hello', 'consumer_tag' => 'consumer', ]; ~~~ **3. 生成者代碼** ~~~ <?php declare (strict_types = 1); namespace app\common\service\rabbitmq; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; /** * rabbitmq 生產者 */ class Producer { // 連接 protected $connection; // 管道 protected $channel; // 配置內容 protected $mqConfig; public function __construct() { $this->mqConfig = config('rabbitmq'); // 創建連接 $this->connection = new AMQPStreamConnection( $this->mqConfig['host'], $this->mqConfig['port'], $this->mqConfig['user'], $this->mqConfig['password'] ); // 創建通道 $this->channel = $this->connection->channel(); } /** * 發送消息 * @param $data 消息內容 */ public function send($data) { /** * 創建隊列(Queue) * name: hello 隊列名稱 * passive: false 如果設置true存在則返回OK，否則就報錯。設置false存在返回OK，不存在則自動創建 * durable: true 是否持久化，設置false是存放到內存中的，RabbitMQ重啟后會丟失;設置true,則代表是一個持久化的隊列，服務重啟后也會存在，因為服務會把持久化的queue存放到磁盤上當服務重啟的時候，會重新加載之前被持久化的queue * exclusive: false 是否排他，指定該選項為true則隊列只對當前連接有效，連接斷開后自動刪除 * auto_delete: false 是否自動刪除，當最后一個消費者斷開連接之后隊列是否自動被刪除 */ $this->channel->queue_declare($this->mqConfig['queue_name'], false, true, false, false); /** * 創建交換機(Exchange) * name: hello 交換機名稱 * type: direct 交換機類型，分別為direct/fanout/topic，參考另外文章的Exchange Type說明。 * passive: false 如果設置true存在則返回OK，否則就報錯。設置false存在返回OK，不存在則自動創建 * durable: false 是否持久化，設置false是存放到內存中的，RabbitMQ重啟后會丟失 * auto_delete: false 是否自動刪除，當最后一個消費者斷開連接之后隊列是否自動被刪除 */ $this->channel->exchange_declare($this->mqConfig['exchange_name'], 'direct', false, true, false); // 綁定消息交換機和隊列 $this->channel->queue_bind($this->mqConfig['queue_name'], $this->mqConfig['exchange_name'],$this->mqConfig['route_key']); // 將要發送數據變為json字符串 $messageBody = json_encode($data, JSON_UNESCAPED_UNICODE); /** * 創建AMQP消息類型 * delivery_mode 消息是否持久化 * AMQPMessage::DELIVERY_MODE_NON_PERSISTENT 不持久化 * AMQPMessage::DELIVERY_MODE_PERSISTENT 持久化 */ $message = new AMQPMessage($messageBody, [ 'content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT ]); /** * 發送消息 * msg: $message AMQP消息內容 * exchange: vckai_exchange 交換機名稱 * routing_key: hello 路由key */ $this->channel->basic_publish($message, $this->mqConfig['exchange_name'], $this->mqConfig['route_key']); // 關閉連接 $this->stop(); } // 關閉連接 public function stop() { $this->channel->close(); $this->connection->close(); } } ~~~ **4. 消費者代碼** ~~~ <?php declare (strict_types = 1); namespace app\common\service\rabbitmq; //use app\common\model\test\MessageModel; use PhpAmqpLib\Connection\AMQPStreamConnection; use think\facade\Log; /** * 消費者 */ class Consumer { // 連接 protected $connection; // 管道 protected $channel; // 配置內容 protected $mqConfig; public function __construct() { $this->mqConfig = config('rabbitmq'); // 創建連接 $this->connection = new AMQPStreamConnection( $this->mqConfig['host'], $this->mqConfig['port'], $this->mqConfig['user'], $this->mqConfig['password'] ); // 創建通道 $this->channel = $this->connection->channel(); } /** * 啟動 * nohup php index.php index/Message_Consume/start & */ public function start() { // 設置消費者（Consumer）客戶端同時只處理一條隊列 // 這樣是告訴RabbitMQ，再同一時刻，不要發送超過1條消息給一個消費者（Consumer）， // 直到它已經處理了上一條消息并且作出了響應。這樣，RabbitMQ就會把消息分發給下一個空閑的消費者（Consumer）。 // 消費者端要把自動確認autoAck設置為false，basic_qos才有效。 // $this->channel->basic_qos(0, 1, false); // 同樣是創建路由和隊列，以及綁定路由隊列，注意要跟producer（生產者）的一致 // 這里其實可以不用設置，但是為了防止隊列沒有被創建所以做的容錯處理 $this->channel->queue_declare($this->mqConfig['queue_name'], false, true, false, false); $this->channel->exchange_declare($this->mqConfig['exchange_name'], 'direct', false, true, false); $this->channel->queue_bind($this->mqConfig['queue_name'], $this->mqConfig['exchange_name'], $this->mqConfig['route_key']); /** * queue: queue_name 被消費的隊列名稱 * consumer_tag: consumer_tag 消費者客戶端身份標識，用于區分多個客戶端 * no_local: false 這個功能屬于AMQP的標準，但是RabbitMQ并沒有做實現 * no_ack: true 收到消息后，是否不需要回復確認即被認為被消費 * exclusive: false 是否排他，即這個隊列只能由一個消費者消費。適用于任務不允許進行并發處理的情況下 * nowait: false 不返回執行結果，但是如果排他開啟的話，則必須需要等待結果的，如果兩個一起開就會報錯 * callback: $callback 回調邏輯處理函數 */ $this->channel->basic_consume($this->mqConfig['queue_name'], $this->mqConfig['consumer_tag'], false, false, false, false, array($this, 'process_message')); register_shutdown_function(array($this, 'shutdown'), $this->channel, $this->connection); while (count($this->channel->callbacks)) { $this->channel->wait(); } } /** * 消息處理 * @param $message */ public function process_message($message) { if ($message->body !== 'quit') { $messageBody = json_decode($message->body); // 自定義的消息類型 if (!isset($messageBody->message_type)) { Log::write("error data:" . $message->body, 2); } else { // $messageModel = new MessageModel(); try { // 消息 Log::write("message_data:" . json_encode($message, JSON_UNESCAPED_UNICODE)); $body = json_decode($message->body, true); dump("消費消息如下："); dump($body); // $messageModel->test($body); } catch (\Think\Exception $e) { Log::write($e->getMessage(), 2); Log::write(json_encode($message), 2); } catch (\PDOException $pe) { Log::write($pe->getMessage(), 2); Log::write(json_encode($message), 2); } } } // 手動確認ack，確保消息已經處理 $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']); // Send a message with the string "quit" to cancel the consumer. if ($message->body === 'quit') { $message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']); } } /** * 關閉進程 * @param $channel * @param $connection */ public function shutdown($channel, $connection) { $channel->close(); $connection->close(); } } ~~~ **5. 創建自定義命令行指令** 在項目跟目錄執行以下命令，會自動生成在 command 目錄生成 Consumer 控制器 ``` php think make:command Consumer ``` 指令內容如下： ~~~ <?php declare (strict_types = 1); namespace app\command; use think\console\Command; use think\console\Input; use think\console\input\Argument; use think\console\input\Option; use think\console\Output; class Consumer extends Command { protected function configure() { // 指令配置 $this->setName('consumer') ->setDescription('the consumer command'); } protected function execute(Input $input, Output $output) { // 啟動消費者 $consumer = new \app\common\service\rabbitmq\Consumer(); $consumer->start(); } } ~~~ config/console.php 代碼如下： ~~~ <?php // +---------------------------------------------------------------------- // | 控制臺配置 // +---------------------------------------------------------------------- return [ // 指令定義 'commands' => [ // rabbitMq 調用消費者 'consumer' => 'app\command\Consumer', ], ]; ~~~ **6. 啟動自定義指令（消費者監聽）** ``` php think consumer ``` **7. 在路由 route/app.php 添加路由并訪問測試** ~~~ Route::get('rabbitmq', function (){ $producer = new \app\common\service\rabbitmq\Producer(); $data = [ 'message_type' => 2, 'order_id' => 3, 'user_id' => 3, 'message' => "發送的消息內容：您的快遞已到的配送站。" ]; $producer->send($data); }); ~~~ ![](https://img.kancloud.cn/e5/ef/e5ef374ab3837ff9fc2ec7cdd913cca1_639x236.png)