php8.0 使用 rabbitmq 要使用 3.6版本以上的， 并且還要開啟 php.ini中的 socket 擴展

php think make:command SimpleMQProduce //創建一個生產者命令行
php think make:command SimpleMQConsumer //創建一個消費者命令行

生產者代碼

<?php
declare (strict_types = 1);namespace app\command;use ba\Exception;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;//本類是模式中的 生產者 produce
class SimpleMQProduce extends Command
{protected function configure(){// 指令配置$this->setName('SimpleMQ')->setDescription('這是一個隊列的 Hello模式 (最簡單的應用模式)，使用了默認的交換機，只需要建一個隊列就可以了');}protected function execute(Input $input, Output $output){//獲取連接$connection = $this->connectRabbitMQ();//獲取連接的通道$channel = $connection->channel();//直接創建一個隊列/*** 關于 queue_declare參數的說明* params  queue  隊列的名稱* params  passive 是否消極的聲明隊列，如果存在，就把隊列的信息返回， 如果沒有就拋出錯誤，（是的， 你沒看錯，這個參數很雞肋，所以一般為 false）* params  durable 是否持久化，意思是說就算隊列服務持了， 也不會丟失隊列* params  exclusive  是否排外，如果設置為true ,表示只有本次連接中的channel 可以訪問，其它channel 是不可以訪問的* params  auto_delete  設置是否自動刪除。為true 則設置隊列為自動刪除。自動刪除的前提是, 至少有一個消費者連接到這個隊列，之后所有與這個隊列連接的消費者都斷開時，才會自動刪除* params  nowait 相當于做一個異步版的聲明， 如果設置成true, 就是說方法調用完就結束，也不用等待創建隊列是否成功，一般也設為false*/$channel->queue_declare("hello",false,false,false,false,false);for ($i = 0; $i < 20; $i++) {$message = ["name"=>"huang".$i,"age"=>$i,"sex"=>"man".$i];$msg = new AMQPMessage(json_encode($message));$channel->basic_publish($msg,"",'hello');  //簡單模式下，routing_key 和 隊列名稱是一樣的sleep(1);}//生產者調用完成后要關閉資源$channel->close();}//這個方法是建立一個 rabbitMQ 的資源連接protected function connectRabbitMQ(){try{$connection = new AMQPStreamConnection("192.168.3.228",'5672',"admin","123456");return $connection;}catch(Exception $e){throw new Exception("隊列連接失敗");}}}

消費者代碼

<?php
declare (strict_types = 1);namespace app\command;use ba\Exception;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;class SimpleMQConsumer extends Command
{protected function configure(){// 指令配置$this->setName('simplemqconsumer')->setDescription('這是rabbitMQ的簡單模式的消費者代碼');}protected function execute(Input $input, Output $output){//創建一個資源連接$connection = $this->connectRabbitMQ();$channel = $connection->channel();//定義一個處理函數$callback = function($msg){$body = $msg->body;$bodyArr = json_decode($body,true);echo $bodyArr["name"] ."--".$bodyArr["age"]. "--".$bodyArr["sex"];};//聲明隊列，這個函數的 passive ，如果有不返回這個隊列的信息，如果沒有就拋出異常$channel->queue_declare("hello",false,false,false,false,false);//從隊列中消費數據$channel->basic_consume("hello",'',false,true,false,false,$callback);//等待隊列執行完成while(count($channel->callbacks)){$channel->wait();}}//這個方法是建立一個 rabbitMQ 的資源連接protected function connectRabbitMQ(){try{$connection = new AMQPStreamConnection("192.168.3.228",'5672',"admin","123456");return $connection;}catch(Exception $e){throw new Exception("隊列連接失敗");}}
}