附錄二安裝RabbitMQ · PHP 教程（付費）

## 好文 * [捫心自問，你真的熟練掌握MQ了嗎](https://mp.weixin.qq.com/s/gqFVeZIE6nfYQd57pQkfaA) ## 使用Docker構建RabbitMQ 查找鏡像 ``` docker search rabbitmq ``` 拉取鏡像 ``` docker pull rabbitmq:3.7.16-management ``` > 默認情況下，會拉取rabbitmq的latest版本。 > 這里拉取 Web瀏覽器管理頁面的tag `3.7.16-management`【管理插件】啟動鏡像 ``` docker run -p 15672:15672 -p 5672:5672 -d --hostname dnmp-rabbitmq --name dnmp-rabbitmq --network dnmp_backend -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin rabbitmq:3.7.16-management ``` > 參數解釋 * `15672` ：表示 RabbitMQ 控制臺端口號，可以在瀏覽器中通過控制臺來執行 RabbitMQ 的相關操作。 * `5672 `: 表示 RabbitMQ 所監聽的 TCP 端口號，應用程序可通過該端口與 RabbitMQ 建立 TCP 連接，完成后續的異步消息通信 * `RABBITMQ_DEFAULT_USER`：用于設置登陸控制臺的用戶名，這里我設置 `admin ` * `RABBITMQ_DEFAULT_PASS`：用于設置登陸控制臺的密碼，這里我設置 `admin `容器啟動成功后，可以在瀏覽器輸入地址：`http://ip:15672/ `訪問控制臺 ## PHP 客戶端庫 > 以下以ThinkPHP5.1 框架為測試環境 #### 安裝擴展庫 [php-amqplib](https://github.com/php-amqplib/php-amqplib) composer installed ``` docker run --rm --interactive --tty -v e:/dnmp/www/iot.tinywan.com:/app composer require php-amqplib/php-amqplib v2.9.0 --ignore-platform-reqs ``` ## 第一章開始代碼 ##### 消息發布者（發送者） > `mq_send.php` 腳本 ``` #!/usr/bin/env php <?php namespace think; define('APP_PATH', __DIR__ . '/application/'); require __DIR__ . '/thinkphp/base.php'; Container::get('app',[APP_PATH])->bind('http/RabbitMq/send')->run()->send(); ``` >[warning] 業務代碼 ``` public function send() { $connection = new AMQPStreamConnection('dnmp-rabbitmq', 5672, 'admin', 'admin'); $channel = $connection->channel(); $channel->queue_declare('hello', false, false, false, false); $msg = new AMQPMessage('Hello World!'); $channel->basic_publish($msg, '', 'hello'); echo " [x] Sent 'Hello World!'\n"; $channel->close(); $connection->close(); } ``` >[danger] 注意： > 1、 `user `和 `password `就是docker啟動時候的 `RABBITMQ_DEFAULT_USER=admin`和 `RABBITMQ_DEFAULT_PASS=admin`。前面我們設置的 `admin`和`admin` > 2、連接主機`host`是`dnmp-rabbitmq`，由于是在docker容器之內 >[info] 在終端中，運行消費者（接收者） ``` > docker exec -it dnmp-php72 sh -c "php /var/www/iot.tinywan.com/mq_receive.php" [*] Waiting for messages. To exit press CTRL+C [x] Received Hello World! [x] Received Hello World! [x] Received Hello World! [x] Received Hello World! ``` ##### 消息接收者（發送者） `mq_receive.php`腳本 ``` #!/usr/bin/env php <?php namespace think; define('APP_PATH', __DIR__ . '/application/'); require __DIR__ . '/thinkphp/base.php'; Container::get('app',[APP_PATH])->bind('http/RabbitMq/receive')->run()->send(); ``` >[warning] 業務代碼 ``` public function receive() { $connection = new AMQPStreamConnection('dnmp-rabbitmq', 5672, 'admin', 'admin'); $channel = $connection->channel(); $channel->queue_declare('hello', false, false, false, false); echo " [*] Waiting for messages. To exit press CTRL+C\n"; $callback = function ($msg) { echo ' [x] Received ', $msg->body, "\n"; }; $channel->basic_consume('hello', '', false, true, false, false, $callback); while (count($channel->callbacks)) { $channel->wait(); } } ``` >[info] 在終端中，運行發布者（發件人） ``` > docker exec -it dnmp-php72 sh -c "php /var/www/iot.tinywan.com/mq_send.php" [x] Sent 'Hello World!' ``` ## 第二章隊列您可能希望看到RabbitMQ有哪些隊列以及它們中有多少消息。您可以使用rabbitmqctl工具（作為特權用戶）執行此操作： ``` docker exec -it dnmp-rabbitmq sh -c "rabbitmqctl list_queues" Timeout: 60.0 seconds ... Listing queues for vhost / ... name messages hello 0 ``` ### 循環調度使用任務隊列的一個優點是能夠輕松地并行工作。如果我們正在積壓工作積壓，我們可以添加更多工人，這樣就可以輕松擴展。首先，讓我們嘗試同時運行兩個`mq_worker.php`腳本。他們都會從隊列中獲取消息，但究竟如何呢？讓我們來看看。你需要打開三個控制臺。兩個將運行`mq_worker.php`腳本。這些游戲機將成為我們的兩個消費者 - `C1`和`C2`。 > 第一個終端 ``` PS E:\dnmp> echo C1 C1 PS E:\dnmp> docker exec -it dnmp-php72 sh -c "php /var/www/iot.tinywan.com/mq_work.php" [*] Waiting for messages. To exit press CTRL+C [x] Received First message [x] Done [x] Received Third message [x] Done [x] Received Fifth message [x] Done ``` > 第二個終端 ``` PS E:\dnmp> echo C2 C2 PS E:\dnmp> docker exec -it dnmp-php72 sh -c "php /var/www/iot.tinywan.com/mq_work.php" [*] Waiting for messages. To exit press CTRL+C [x] Received First message [x] Done [x] Received Third message [x] Done [x] Received Fifth message [x] Done ``` >[warning] 第三個終端我們將發布新任務。啟動消費者后，您可以發布一些消息： ``` docker exec -it dnmp-php72 sh -c "php /var/www/iot.tinywan.com/mq_new_task.php First message. " docker exec -it dnmp-php72 sh -c "php /var/www/iot.tinywan.com/mq_new_task.php Second message. " docker exec -it dnmp-php72 sh -c "php /var/www/iot.tinywan.com/mq_new_task.php Third message. " docker exec -it dnmp-php72 sh -c "php /var/www/iot.tinywan.com/mq_new_task.php Fourth message. " docker exec -it dnmp-php72 sh -c "php /var/www/iot.tinywan.com/mq_new_task.php Fifth message. " ``` #### 查看其他兩個worker是如何接受消息的 ``` > docker exec -it dnmp-php72 sh -c "php /var/www/iot.tinywan.com/mq_work.php" [*] Waiting for messages. To exit press CTRL+C [x] Received First message [x] Done [x] Received Third message [x] Done [x] Received Fifth message [x] Done ``` ``` > docker exec -it dnmp-php72 sh -c "php /var/www/iot.tinywan.com/mq_work.php" [*] Waiting for messages. To exit press CTRL+C [x] Received Second message [x] Done [x] Received Fourth message [x] Done ``` > 默認情況下，RabbitMQ將按順序將每條消息發送給下一個消費者。平均而言，每個消費者將獲得相同數量的消息。這種分發消息的方式稱為循環法。與三個或更多工人一起嘗試。 ### 消息確認執行任務可能需要幾秒鐘。你可能想知道如果其中一個消費者開始一項長期任務并且只是部分完成而死亡會發生什么。使用我們當前的代碼，一旦RabbitMQ向消費者發送消息，它立即將其標記為刪除。在這種情況下，如果你殺死一個工人，我們將丟失它剛剛處理的消息。我們還將丟失分發給這個特定工作者但尚未處理的所有消息。但我們不想失去任何任務。如果工人死亡，我們希望將任務交付給另一名工人。為了確保消息永不丟失，RabbitMQ支持[消息*確認*](https://www.rabbitmq.com/confirms.html)。消費者發回ack（nowledgement）告訴RabbitMQ已收到，處理了特定消息，RabbitMQ可以自由刪除它。如果消費者死亡（其通道關閉，連接關閉或TCP連接丟失）而不發送確認，RabbitMQ將理解消息未完全處理并將重新排隊。如果同時有其他在線消費者，則會迅速將其重新發送給其他消費者。這樣你就可以確保沒有消息丟失，即使工人偶爾會死亡。沒有任何消息超時;當消費者死亡時，RabbitMQ將重新發送消息。即使處理消息需要非常長的時間，也沒關系。默認情況下，消息確認已關閉。現在是時候通過設置第四個參數來打開它們`basic_consume`到假（`true`表示*`沒有ACK*`），并從工作人員發送適當的確認，一旦我們有一個任務來完成。 #### `即將被消費`和`未被消費`的消息查看 ``` > docker exec -it dnmp-rabbitmq sh -c "rabbitmqctl list_queues name messages_ready messages_unacknowledged" Timeout: 60.0 seconds ... Listing queues for vhost / ... name messages_ready messages_unacknowledged task_queue 4 0 hello 0 0 ``` > `messages_ready `：即將被消費 > `messages_unacknowledged `：消費但是未被確認 > 這時候如果開啟一個work，則會全部被消費掉 ``` > docker exec -it dnmp-php72 sh -c "php /var/www/iot.tinywan.com/mq_work.php" [*] Waiting for messages. To exit press CTRL+C [x] Received Fifth message [x] Done [x] Received Fifth message [x] Done [x] Received Fifth message [x] Done [x] Received Fifth message [x] Done ``` 再次查看，準備的消息已經全部被消費完了 ``` > docker exec -it dnmp-rabbitmq sh -c "rabbitmqctl list_queues name messages_ready messages_unacknowledged" Timeout: 60.0 seconds ... Listing queues for vhost / ... name messages_ready messages_unacknowledged task_queue 0 0 hello 0 0 ``` ### 消息持久性我們已經學會了如何確保即使消費者死亡，任務也不會丟失。但是如果RabbitMQ服務器停止，我們的任務仍然會丟失。當RabbitMQ退出或崩潰時，它將忘記隊列和消息，除非你告訴它不要。確保消息不會丟失需要做兩件事：我們需要將隊列和消息都標記為持久。首先，我們需要確保RabbitMQ永遠不會丟失我們的隊列。為此，我們需要聲明它是持久的。為此，我們將第三個參數傳遞給`queue_declare`為`true`： ``` $channel->queue_declare('hello', false, true, false, false); ``` 這里如果實驗 1、關閉RabbitMQ服務器：`docker stop dnmp-rabbitmq` 2、重啟RabbitMQ服務器：`docker start dnmp-rabbitmq` 3、重新查看消息同時消費消息，消息是否已經持久化在數據文件中`rabbitmqctl list_queues name messages_ready messages_unacknowledged` 4、開啟一個worker 消費消息，看看消息是否還在 ## 第三章發布訂閱 #### 交易所讓我們快速瀏覽前面教程中介紹的內容： * 甲*生產者*是發送消息的用戶的應用程序。 * 甲*隊列*是存儲消息的緩沖器。 * 甲*消費者*是接收消息的用戶的應用程序。 RabbitMQ中消息傳遞模型的核心思想是生產者永遠不會將任何消息直接發送到隊列。實際上，生產者通常甚至不知道消息是否會被傳遞到任何隊列。