thinkphp+rabbitmq · 編程知識

**1、安裝rabbitmq** **2、crontab定時檢測rabbtimq狀態** **2、使用thinphp6.0框架rabbitmq示例,supervisor守護消費者** **3、RabbitMQ有四種交換機類型** ![](https://img.kancloud.cn/fd/4d/fd4d3c3a504c4aba3d9aac0aeacd71b2_1843x781.png) [rabbitmq](https://so.csdn.net/so/search?q=rabbitmq&spm=1001.2101.3001.7020 "rabbitmq")組成部分如下： > Broker：消息隊列服務進程，此進程包括兩個部分：Exchange和Queue。 > Exchange：消息隊列交換機，按一定的規則將消息路由轉發到某個隊列對消息進行過慮。 > Queue：消息隊列，存儲消息的隊列，消息到達隊列并轉發給指定的消費方。 > Producer：消息生產者，即生產方客戶端，生產方客戶端將消息發送到MQ。 > Consumer：消息消費者，即消費方客戶端，接收MQ轉發的消息。消息發布接收流程： > ?—–發送消息—– > 1、生產者和Broker建立TCP連接。 > 2、生產者和Broker建立通道。 > 3、生產者通過通道消息發送給Broker，由Exchange將消息進行轉發。 > 4、Exchange將消息轉發到指定的Queue（隊列） ?—-接收消息—– > 1、消費者和Broker建立TCP連接 > 2、消費者和Broker建立通道 > 3、消費者監聽指定的Queue（隊列） > 4、當有消息到達Queue時Broker默認將消息推送給消費者。 > 5、消費者接收到消息。 > 6、消費者應使用[supervisor](https://so.csdn.net/so/search?q=supervisor&spm=1001.2101.3001.7020)進行監控保持在線 ![](https://img.kancloud.cn/29/18/291802e9ca5851273a3d3f6ba15a14e0_521x429.png) ### **1、方式一：docker安裝RabbitMQ** **查看倉庫里的RabbitMQ** ~~~csharp [root@localhost ~]# docker search rabbitmq ~~~ ![](https://img-blog.csdnimg.cn/img_convert/d7cb05791809d5c58853cca17fd5ff54.png) ?**安裝RabbitMQ** ~~~csharp [root@localhost ~]# docker pull rabbitmq ~~~ 這里是直接安裝最新的，如果需要安裝其他版本在[rabbitmq](https://so.csdn.net/so/search?q=rabbitmq&spm=1001.2101.3001.7020)后面跟上版本號即可 **啟動RabbitMQ** ~~~cobol [root@localhost ~]# docker run -d --hostname my-rabbit --name rabbit -p 15672:15672 -p 5672:5672 rabbitmq ~~~ **安裝插件** 1\. 先執行docker ps 拿到當前的鏡像ID ~~~csharp [root@localhost ~]# docker ps ~~~ 2\. 進入容器 ~~~csharp [root@localhost ~]# docker exec -it 鏡像ID /bin/bash ~~~ 3\. 安裝web頁面插件 ~~~csharp [root@localhost ~]# rabbitmq-plugins enable rabbitmq_management ~~~ ### 2、方式二：[CentOS6.9下安裝rabbitmq消息隊列](https://link.zhihu.com/?target=https%3A//www.cnblogs.com/vipzhou/p/7890674.html "CentOS6.9下安裝rabbitmq消息隊列") 安裝如下步驟： **首先安裝erlang** ~~~csharp [root@localhost ~]# yum install erlang ~~~ **安裝rabbitmq rpm包** ~~~cobol [root@localhost ~]# wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.5.0/rabbitmq-server-3.5.0-1.noarch.rpm[root@localhost ~]# rpm -ivh rabbitmq-server-3.5.0-1.noarch.rpm ~~~ **啟動rabbitmq** ~~~csharp [root@localhost ~]# service start rabbitmq-server ~~~ 注：如果啟動失敗，報錯如下： ![](https://img-blog.csdnimg.cn/img_convert/d10d887afc3fe069cbd000b72a9c473b.png) ?則修改hosts，添加hostname： ~~~csharp [root@localhost ~]# vi /etc/hosts ~~~ 添加hostname： ![](https://img-blog.csdnimg.cn/img_convert/55255967e515e5444d49c74a03b80adf.png) ?再次重啟即可： ![](https://img-blog.csdnimg.cn/img_convert/b52a94300cff7730be98bbab6943a844.png) ?**打開5672端口（**注意 15672，5672端口防火墻之類的問題需要開啟一下**）** ~~~cobol [root@localhost ~]# /sbin/iptables -I INPUT -p tcp --dport 5672 -j ACCEPT [root@localhost ~]# /etc/rc.d/init.d/iptables save [root@localhost ~]# /etc/init.d/iptables restart ~~~ ![](https://img-blog.csdnimg.cn/img_convert/5992d0ce634f4201c29c774966bf7cd0.png) ?**啟用web頁面插件** ~~~csharp [root@localhost ~]# rabbitmq-plugins enable rabbitmq_management ~~~ **重啟rabbitmq** ~~~csharp [root@localhost ~]# systemctl start rabbitmq-server ~~~ **web訪問：**?ip地址**可以通過ip add 或者 ifconfig查看** **用戶名和密碼：guest guest** **![](https://img-blog.csdnimg.cn/img_convert/c69ec3b74b754ede7103e976ee61fb0d.png)** ?**無法登錄請使用如下：** ~~~cobol [root@localhost ~]# vim /etc/rabbitmq/rabbitmq.config ~~~ 寫入如下信息并保存： ~~~less [{rabbit, [{loopback_users, []}]}]. ~~~ ![](https://img-blog.csdnimg.cn/img_convert/04418c573b9b16ff5a7dc0f333856709.png) ?再重啟： ![](https://img-blog.csdnimg.cn/img_convert/7b6fb9f69f39fb571f6d8136a00b9786.png) ?再次訪問即可： ![](https://img-blog.csdnimg.cn/img_convert/e63f036cbbc828a8d78420aee6ba8628.webp?x-oss-process=image/format,png) > ?服務啟動 > \[root@localhost ~\]# systemctl start rabbitmq-server.service > 看看是否啟動成功 > \[root@localhost ~\]# rabbitmqctl status > 設置開機自啟 > \[root@localhost ~\]# \[root@localhost ~\]# chkconfig rabbitmq-server on > 添加到啟動項并設置開機自啟 > \[root@localhost ~\]# systemctl enable rabbitmq-server.service > 開啟管理界面 > \[root@localhost ~\]# rabbitmq-plugins enable rabbitmq\_management > 添加賬號 > \[root@localhost ~\]# rabbitmqctl add\_user abc 123456 > 設置用戶角色 > \[root@localhost ~\]# rabbitmqctl set\_user\_tags abc administrator > 設置用戶權限 > \[root@localhost ~\]# rabbitmqctl set\_permissions -p "/" abc ".\*" ".\*" ".\*" > 查看用戶和角色需要啟動服務 > \[root@localhost ~\]# rabbitmqctl list\_users > 刪除角色 > \[root@localhost ~\]# rabbitmqctl delete\_user Username **2、crontab定時檢測rabbtimq狀態（應使用supervisor實時監聽）** 1）[crontab](https://so.csdn.net/so/search?q=crontab&spm=1001.2101.3001.7020 "crontab")定時任務(每分鐘檢查運行腳本【分，時，日，月，周】) \[root@localhost ~\]#[crontab](https://so.csdn.net/so/search?q=crontab&spm=1001.2101.3001.7020)\-e ~~~cobol ...*/1 * * * * /bin/bash /usr/local/sbin/rabbitmq.sh ~~~ ?2）rabbitmq啟動腳本（通過獲取pgrep獲取rabbitmq-server進程號再進行判斷） ~~~cobol [root@localhost ~]# cd /usr/local/sbin[root@localhost ~]# vim rabbitmq.sh#!/bin/bash pgrep -x rabbitmq-server &> /dev/null if [ $? -ne 0 ] then echo "At time: `date` :rabbitmq error stop .">> /var/log/rabbitmqCheck.log/etc/init.d/rabbitmq-server start#echo "At time: `date` :rabbitmq server is stop." elseecho "rabbitmq server is running ." >> /var/log/rabbitmqCheck.log fi ~~~ 測試： 1）停止rabbitmq【Active:inactive(dead)】 ~~~cobol [root@localhost sbin]# /etc/init.d/rabbitmq-server stopStopping rabbitmq-server (via systemctl): [ OK ][root@localhost sbin]# [root@localhost sbin]# systemctl status rabbitmq-server● rabbitmq-server.service - LSB: Enable AMQP service provided by RabbitMQ broker Loaded: loaded (/etc/rc.d/init.d/rabbitmq-server; bad; vendor preset: disabled) Active: inactive (dead) since Wed 2021-01-13 18:01:33 CST; 46s ago Docs: man:systemd-sysv-generator(8) Process: 14618 ExecStop=/etc/rc.d/init.d/rabbitmq-server stop (code=exited, status=0/SUCCESS) Process: 14346 ExecStart=/etc/rc.d/init.d/rabbitmq-server start (code=exited, status=0/SUCCESS) Jan 13 18:01:02 localhost.localdomain su[14426]: (to rabbitmq) root on noneJan 13 18:01:02 localhost.localdomain su[14421]: (to rabbitmq) root on noneJan 13 18:01:05 localhost.localdomain rabbitmq-server[14346]: Starting rabbitmq-server: SUCCESSJan 13 18:01:05 localhost.localdomain rabbitmq-server[14346]: rabbitmq-server.Jan 13 18:01:05 localhost.localdomain systemd[1]: Started LSB: Enable AMQP service provided by RabbitMQ broker.Jan 13 18:01:29 localhost.localdomain systemd[1]: Stopping LSB: Enable AMQP service provided by RabbitMQ broker...Jan 13 18:01:29 localhost.localdomain su[14623]: (to rabbitmq) root on noneJan 13 18:01:30 localhost.localdomain su[14684]: (to rabbitmq) root on noneJan 13 18:01:33 localhost.localdomain rabbitmq-server[14618]: Stopping rabbitmq-server: rabbitmq-server.Jan 13 18:01:33 localhost.localdomain systemd[1]: Stopped LSB: Enable AMQP service provided by RabbitMQ broker.[root@localhost sbin]# ~~~ 2）隔一分鐘再次查看rabbitmq狀態【Active:active running】 \[root@localhost sbin\]# systemctl status rabbitmq-server ~~~cobol ● rabbitmq-server.service - LSB: Enable AMQP service provided by RabbitMQ broker Loaded: loaded (/etc/rc.d/init.d/rabbitmq-server; bad; vendor preset: disabled) Active: active (running) since Wed 2021-01-13 18:03:04 CST; 35s ago Docs: man:systemd-sysv-generator(8) Process: 14618 ExecStop=/etc/rc.d/init.d/rabbitmq-server stop (code=exited, status=0/SUCCESS) Process: 14917 ExecStart=/etc/rc.d/init.d/rabbitmq-server start (code=exited, status=0/SUCCESS) Tasks: 2 Memory: 1.4M CGroup: /system.slice/rabbitmq-server.service ├─14990 /bin/sh /etc/rc.d/init.d/rabbitmq-server start └─14995 /bin/bash -c ulimit -S -c 0 >/dev/null 2>&1 ; /usr/sbin/rabbitmq-server Jan 13 18:03:01 localhost.localdomain systemd[1]: Starting LSB: Enable AMQP service provided by RabbitMQ broker...Jan 13 18:03:01 localhost.localdomain su[14921]: (to rabbitmq) root on noneJan 13 18:03:02 localhost.localdomain su[14998]: (to rabbitmq) root on noneJan 13 18:03:02 localhost.localdomain su[14991]: (to rabbitmq) root on noneJan 13 18:03:04 localhost.localdomain rabbitmq-server[14917]: Starting rabbitmq-server: SUCCESSJan 13 18:03:04 localhost.localdomain systemd[1]: Started LSB: Enable AMQP service provided by RabbitMQ broker.Jan 13 18:03:04 localhost.localdomain rabbitmq-server[14917]: rabbitmq-server.You have new mail in /var/spool/mail/root[root@localhost sbin]# ~~~ ### 方式三：docker-compose.yml安裝 ![](https://img-blog.csdnimg.cn/3cf5516116ee482f973fa6ffc3b9c61e.png) ~~~cobol version: '3.1'services: rabbitmq: hostname: rabbitmq image: rabbitmq:3.9.12-management container_name: rabbitmq privileged: true ports: - 15672:15672 - 5672:5672 volumes: - /etc/localtime:/etc/localtime - ./data:/var/lib/rabbitmq environment: TZ: Asia/Shanghai RABBITMQ_DEFAULT_USER: admin RABBITMQ_DEFAULT_PASS: 123456 ~~~ ### **3、thinkphp6.0使用composer安裝rabbitmq安裝包** **本次需要修改的文件有** ~~~delphi Comer.php 使用php think生產的命令行Index.php 測試類MqConsumer.php 消費者MqProducer.php 生產者Console.php 命令行配置項rabbitmq.php 消息隊列配置項 ~~~ ![](https://img-blog.csdnimg.cn/img_convert/3820aa96330efb71bf405ddeb4c608f2.webp?x-oss-process=image/format,png) **1、安裝amqplib** 進入到tp6項目根目錄安裝rabbitmq包，如需忽略版本安裝 --ignore-platform-reqs ~~~php $composer require --ignore-platform-reqs php-amqplib/php-amqplib ~~~ ?![](https://img-blog.csdnimg.cn/img_convert/a7f7a1c8dcf1359a1001dc741e4b93b0.png) **2、在config文件夾下添加rabbitmq.php配置文件** ~~~cobol <?php// 示例配置文件return [ # 連接信息 'AMQP' => [ 'host' => '192.168.1.130', //連接rabbitmq,此為安裝rabbitmq服務器 'port'=>'5672', 'login'=>'guest', 'password'=>'guest', 'vhost'=>'/' ], # 隊列 'direct_queue' => [ 'exchange_name' => 'direct_exchange', 'exchange_type'=>'direct',#直連模式 'queue_name' => 'direct_queue', 'route_key' => 'direct_roteking', 'consumer_tag' => 'direct' ]]; ~~~ ?**3、編寫生產者代碼** ~~~cobol <?phpnamespace app\controller; use app\BaseController;use PhpAmqpLib\Connection\AMQPStreamConnection;use PhpAmqpLib\Message\AMQPMessage;use think\facade\Log; class MqProducer{ public static function pushMessage($data) { $param = config('rabbitmq.AMQP'); $amqpDetail = config('rabbitmq.direct_queue'); $connection = new AMQPStreamConnection( $param['host'], $param['port'], $param['login'], $param['password'], $param['vhost'] ); $channel = $connection->channel(); /* * 創建隊列(Queue) * name: hello // 隊列名稱 * passive: false // 如果設置true存在則返回OK，否則就報錯。設置false存在返回OK，不存在則自動創建 * durable: true // 是否持久化，設置false是存放到內存中RabbitMQ重啟后會丟失, * 設置true則代表是一個持久的隊列，服務重啟之后也會存在，因為服務會把持久化的Queue存放在硬盤上，當服務重啟的時候，會重新加載之前被持久化的Queue * exclusive: false // 是否排他，指定該選項為true則隊列只對當前連接有效，連接斷開后自動刪除 * auto_delete: false // 是否自動刪除，當最后一個消費者斷開連接之后隊列是否自動被刪除 */ $channel->queue_declare($amqpDetail['queue_name'], false, true, false, false); /* * 創建交換機(Exchange) * name: vckai_exchange// 交換機名稱 * type: direct // 交換機類型，分別為direct/fanout/topic，參考另外文章的Exchange Type說明。 * passive: false // 如果設置true存在則返回OK，否則就報錯。設置false存在返回OK，不存在則自動創建 * durable: false // 是否持久化，設置false是存放到內存中的，RabbitMQ重啟后會丟失 * auto_delete: false // 是否自動刪除，當最后一個消費者斷開連接之后隊列是否自動被刪除 */ $channel->exchange_declare($amqpDetail['exchange_name'], $amqpDetail['exchange_type'], false, true, false); /* * 綁定隊列和交換機 * @param string $queue 隊列名稱 * @param string $exchange 交換器名稱 * @param string $routing_key 路由key * @param bool $nowait * @param array $arguments * @param int|null $ticket * @throws \PhpAmqpLib\Exception\AMQPTimeoutException if the specified operation timeout was exceeded * @return mixed|null */ $channel->queue_bind($amqpDetail['queue_name'], $amqpDetail['exchange_name'], $amqpDetail['route_key']); /* $messageBody:消息體 content_type:消息的類型可以不指定 delivery_mode:消息持久化最關鍵的參數 AMQPMessage::DELIVERY_MODE_NON_PERSISTENT = 1; 不持久化 AMQPMessage::DELIVERY_MODE_PERSISTENT = 2; 持久化 */ //將要發送數據變為json字符串 $messageBody = json_encode($data); /* * 創建AMQP消息類型 * $messageBody:消息體 * delivery_mode 消息是否持久化 * AMQPMessage::DELIVERY_MODE_NON_PERSISTENT = 1; 不持久化 * AMQPMessage::DELIVERY_MODE_PERSISTENT = 2; 持久化 */ $message = new AMQPMessage($messageBody, array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT)); /* * 發送消息 * msg // AMQP消息內容 * exchange // 交換機名稱 * routing key // 路由鍵名稱 */ $channel->basic_publish($message, $amqpDetail['exchange_name'],$amqpDetail['route_key']); $channel->close(); $connection->close(); echo "ok"; }} ~~~ **4、消費者代碼** ~~~cobol <?phpnamespace app\controller; use app\BaseController;use PhpAmqpLib\Connection\AMQPStreamConnection;use think\Controller;use think\facade\Log; class MqConsumer{ /** * 消費端消費端需要保持運行狀態實現方式 * 1 linux上寫定時任務每隔5分鐘運行下該腳本，保證訪問服務器的ip比較平緩，不至于崩潰 * 2 nohup php index.php index/Message_Consume/start & 用nohup命令后臺運行該腳本 * 3 **/ function shutdown($channel, $connection) { $channel->close(); $connection->close(); Log::write("closed",3); } //消息處理 function process_message($message) { //休眠兩秒 //sleep(2); echo $message->body."\n"; //自定義日志為rabbitmq-consumer Log::write($message->body,'rabbitmq-consumer'); //[2021-01-14T16:14:17+08:00][rabbitmq-consumer] {"time":1610612057,"order":85} //手動發送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']); } } /** * 啟動 * @return \think\Response */ public function start() { $param = config('rabbitmq.AMQP'); $amqpDetail = config('rabbitmq.direct_queue'); $connection = new AMQPStreamConnection( $param['host'], $param['port'], $param['login'], $param['password'], $param['vhost'] ); /* * 創建通道 */ $channel = $connection->channel(); /* * 設置消費者（Consumer）客戶端同時只處理一條隊列 * 這樣是告訴RabbitMQ，再同一時刻，不要發送超過1條消息給一個消費者（Consumer）， * 直到它已經處理了上一條消息并且作出了響應。這樣，RabbitMQ就會把消息分發給下一個空閑的消費者（Consumer）。 */ $channel->basic_qos(0, 1, false); /* * 同樣是創建路由和隊列，以及綁定路由隊列，注意要跟publisher的一致 * 這里其實可以不用，但是為了防止隊列沒有被創建所以做的容錯處理 */ $channel->queue_declare($amqpDetail['queue_name'], false, true, false, false); $channel->exchange_declare($amqpDetail['exchange_name'], $amqpDetail['exchange_type'], false, true, false); $channel->queue_bind($amqpDetail['queue_name'], $amqpDetail['exchange_name'],$amqpDetail['route_key']); /* queue: 從哪里獲取消息的隊列 consumer_tag: 消費者標識符,用于區分多個客戶端 no_local: 不接收此使用者發布的消息 no_ack: 設置為true，則使用者將使用自動確認模式。詳情請參見. 自動ACK：消息一旦被接收，消費者自動發送ACK 手動ACK：消息接收后，不會發送ACK，需要手動調用 exclusive:是否排他，即這個隊列只能由一個消費者消費。適用于任務不允許進行并發處理的情況下 nowait: 不返回執行結果，但是如果排他開啟的話，則必須需要等待結果的，如果兩個一起開就會報錯 callback: :回調邏輯處理函數,PHP回調 array($this, 'process_message') 調用本對象的process_message方法 */ $channel->basic_consume($amqpDetail['queue_name'], $amqpDetail['consumer_tag'], false, false, false, false, array($this, 'process_message')); register_shutdown_function(array($this, 'shutdown'), $channel, $connection); // 阻塞隊列監聽事件 while (count($channel->callbacks)) { $channel->wait(); } }} ~~~ **5、編寫測試代碼Index.php** ~~~cobol <?phpnamespace app\controller; use app\BaseController;use app\controller\MqProducer; class Index{ public function index() { echo "tp6";die; } public function hello($name = 'ThinkPHP6') { return 'hello,' . $name; } public function send() { // for($i=0; $i<5; $i++){ $consumer = new MqProducer();//生產者 $data = [ 'time'=>time(), 'order'=> rand(1, 100), ]; $consumer->pushMessage($data); // sleep(1);// } }} ~~~ 6、項目更目錄使用php think生成指令 1)進入到項目目錄：php -v 查看是否配置環境，否則配置php系統環境 Windows7下的php環境配置教程\_php技巧\_腳本之家?www.jb51.net/article/61507.htm正在上傳…重新上傳取消 2)執行：php think make:command Comer Comer 會在command目錄下生成Comer .php文件 ![](https://img-blog.csdnimg.cn/img_convert/21c9c386fefa84187404ddd54aaae8ca.webp?x-oss-process=image/format,png) 3)在config/console.php添加指令 ~~~php <?php// +----------------------------------------------------------------------// | 控制臺配置// +----------------------------------------------------------------------return [ // 指令定義 'commands' => [ // consumer是app\Command\Consumer文件中自定義命令行的名字不能使用 'comer' => 'app\command\Comer', ],]; ~~~ ?4.修改指令文件 ~~~php <?phpdeclare (strict_types = 1); namespace app\command; use app\controller\MqConsumer;use think\console\Command;use think\console\Input;use think\console\input\Argument;use think\console\input\Option;use think\console\Output; class Comer extends Command{ protected function configure() { // 指令配置 $this->setName('Comer') ->setDescription('the Comer command'); } protected function execute(Input $input, Output $output) { // 指令輸出// $output->writeln('Comer'); $consumer = new MqConsumer(); //消費者 $consumer->start(); //啟動 }} ~~~ **7、測試** 步驟： > （1）啟動監聽指令（消費者消費） > （2）生產者發送消息 > （3）查看rabbitmq管理界面是否有消息 > （4）查看消費者已消費消息 **1)在項目根目錄指令監聽 php think comer** **![](https://img-blog.csdnimg.cn/img_convert/6a305eeb9f3b7e54a1a366c807f25798.png)**? ?**2）發送消息** ~~~php http://www.tp6.net/index.php/index/send ~~~ ![](https://img-blog.csdnimg.cn/img_convert/926d9ef3f5380f24a04da97eb8fb04e8.png) **3）rabbitmq管理界面** **![](https://img-blog.csdnimg.cn/img_convert/755d2ca507971f915ffbd40011887587.png)**? **4）查看消費者已消費消息** **![](https://img-blog.csdnimg.cn/img_convert/9cee6605d5c36a07f0f4901e39b24094.webp?x-oss-process=image/format,png)**? **8、使用supervisor監控消費者實時在線** **安裝supervisor:** 監聽守護【rabbitmq消費者】配置文件如下,然后保存退出 > \[user@localhost??[tp6.com](https://link.zhihu.com/?target=http%3A//www.tp6.com/ "tp6.com")\]$ cd /etc/supervisord.d/ > \[user@localhost supervisord.d\]$ sudo vim tp\_amqp.ini ~~~php [program:tp_amqp]directory = /www/wwwroot/www.tp6.com ;啟動目錄command = /www/server/php/72/bin/php think comer ;啟動命令autostart = true ;在supervisord啟動的時候也啟動startsecs = 5 ;啟動5秒后沒有異常退出，就當作已經正常啟動了autorestart = true ;程序異常退出后自動重啟startretries = 3 ;啟動失敗自動重試次數，默認是3user = root ;哪個用戶啟動redirect_stderr = true ;把stderr重定向到stdout，默認falsestdout_logfile_maxbytes = 20MB ;stdout日志文件大小，默認50MBstdout_logfile_backups = 20 ;stdout日志文件備份數stdout_logfile = /usr/log/tp_amqp.log;stdout日志文件，需要手動創建/usr/log/tp_amqp.log ~~~ ?**4、啟動守護進程** > 更新配置文件：supervisorctl update > 啟動進程：sudo supervisorctl start tp\_amqp 通過ps查看守護的命令： ~~~php [user@localhost www.tp6.com]$ sudo supervisorctl status tp_amqptp_amqp RUNNING pid 16595, uptime 0:01:27[user@localhost www.tp6.com]$ ps afx | grep php 2364 ? Ss 0:00 php-fpm: master process (/www/server/php/72/etc/php-fpm.conf) 2365 ? S 0:00 \_ php-fpm: pool www 2366 ? S 0:00 \_ php-fpm: pool www 2367 ? S 0:00 \_ php-fpm: pool www 2368 ? S 0:00 \_ php-fpm: pool www 2369 ? S 0:00 \_ php-fpm: pool www14478 ? S 0:00 \_ php-fpm: pool www16746 pts/1 S+ 0:00 \_ grep --color=auto php16595 ? S 0:00 \_ /www/server/php/72/bin/php think comer[user@localhost www.tp6.com]$ sudo kill 16595[user@localhost www.tp6.com]$ ps afx | grep php 2364 ? Ss 0:00 php-fpm: master process (/www/server/php/72/etc/php-fpm.conf) 2365 ? S 0:00 \_ php-fpm: pool www 2366 ? S 0:00 \_ php-fpm: pool www 2367 ? S 0:00 \_ php-fpm: pool www 2368 ? S 0:00 \_ php-fpm: pool www 2369 ? S 0:00 \_ php-fpm: pool www14478 ? S 0:00 \_ php-fpm: pool www16823 pts/1 S+ 0:00 \_ grep --color=auto php16821 ? S 0:00 \_ /www/server/php/72/bin/php think comer[user@localhost www.tp6.com]$ ~~~ 簡單描述一下RabbitMQ中的幾個關鍵的概念： Broker：可以簡單的理解為安裝了RabbitMQ服務的這臺機器就可以稱為中間人 Exchange：交換機，消息經由它，通過路由鍵來判斷并決定把消息投遞給哪個隊列，它類似于一個路由器的角色 Queue：隊列，最終將消息投遞到隊列中，由消費端監聽隊列進行消費 Binding：綁定關系，需要給交換機綁定隊列，綁定時需要給一個路由鍵 Routingkey：路由鍵，交換機和隊列進行綁定時，需要指定路由鍵或通配符路由鍵。交換機根據路由鍵來決定消息投遞到哪個或哪些隊列大致流程：使用RabbitMQ前，首先需要根據業務來創建交換機和隊列，創建完成后需要給交換機綁定隊列（交換機和隊列可以是多對多的關系），綁定隊列時要指定具體的路由鍵或者通配符路由鍵當生產者發送一條消息的時候，需要指定交換機和路由鍵，消息到達Broker后先轉給剛才指定的交換機，交換機再根據路由鍵來決定把消息投遞給與自己綁定的哪一個或哪一些隊列，最后再由消費端來監聽這些隊列，消費處理對應的消息最新版本的RabbitMQ有四種交換機類型，分別是：Direct exchange、Fanout exchange、Topic exchange、Headers exchange #### 1、Direct exchange---直接類型交換機要求消息帶的路由鍵和綁定的路由鍵**完全**匹配，這是一個完整的匹配。\\ ![](https://img-blog.csdnimg.cn/img_convert/0d35df42391a77904ead445550688bda.png) 2、Fanout Exchange---扇出類型交換機只需要簡單的將隊列綁定到該類型交換機上，該類型的交換機綁定隊列時可以不指定路由鍵(Routingkey) 當消息發送給該交換機后，它會將消息投遞給與該交換機綁定的所有隊列很像廣播，每臺子網內的機器都會獲得一份消息，Fanout交換機轉發消息是最快的 ![](https://img-blog.csdnimg.cn/img_convert/8b0181c5e2447fd6a08ced06aa8a8bbb.png) #### 3、Topic Exchange---主題類型交換機將路由鍵和某模式進行匹配。此時隊列需要綁定某一個模式上。符號#匹配0個或多個單詞，符號 \*匹配一個單詞。 ![](https://img-blog.csdnimg.cn/img_convert/2aee2ec45669b7077cb4fe69797c1d50.webp?x-oss-process=image/format,png) #### 4、Headers Exchanges?