一、 名詞解釋
-
Connection、Channel
Connection、Channel都是RabbitMQ對外提供的API中最基本的對象。
Connection是RabbitMQ的socket鏈接，它封裝了socket協議相關部分邏輯。
Channel是我們與RabbitMQ打交道的最重要的一個接口，我們大部分的業務操作是在Channel這個接口中完成的，包括定義Queue、定義Exchange、綁定Queue與Exchange、發布消息等。 -
Queue:
Queue（隊列）是RabbitMQ的內部對象，用於存儲消息，用下圖表示。
RabbitMQ中的消息都只能存儲在Queue中，生產者（下圖中的P）生產消息並最終投遞到Queue中，消費者（下圖中的C）可以從Queue中獲取消息並消費
多個消費者可以訂閱同一個Queue，這時Queue中的消息會被平均分攤給多個消費者進行處理，而不是每個消費者都收到所有的消息並處理。 -
Message acknowledgment
在實際應用中，可能會發生消費者收到Queue中的消息，但沒有處理完成就宕機（或出現其他意外）的情況，這種情況下就可能會導致消息丟失。為了避免這種情況發生，我們可以要求消費者在消費完消息后發送一個回執給RabbitMQ，RabbitMQ收到消息回執（Message acknowledgment）后才將該消息從Queue中移除；如果RabbitMQ沒有收到回執並檢測到消費者的RabbitMQ連接斷開，則RabbitMQ會將該消息發送給其他消費者（如果存在多個消費者）進行處理。這里不存在timeout概念，一個消費者處理消息時間再長也不會導致該消息被發送給其他消費者，除非它的RabbitMQ連接斷開。
這里會產生另外一個問題，如果我們的開發人員在處理完業務邏輯后，忘記發送回執給RabbitMQ，這將會導致嚴重的bug——Queue中堆積的消息會越來越多；消費者重啟后會重復消費這些消息並重復執行業務邏輯… -
Message durability
如果我們希望即使在RabbitMQ服務重啟的情況下，也不會丟失消息，我們可以將Queue與Message都設置為可持久化的（durable），這樣可以保證絕大部分情況下我們的RabbitMQ消息不會丟失。但依然解決不了小概率丟失事件的發生（比如RabbitMQ服務器已經接收到生產者的消息，但還沒來得及持久化該消息時RabbitMQ服務器就斷電了），如果我們需要對這種小概率事件也要管理起來，那么我們要用到事務。由於這里僅為RabbitMQ的簡單介紹，所以這里將不講解RabbitMQ相關的事務。
-
Prefetch count
前面我們講到如果有多個消費者同時訂閱同一個Queue中的消息，Queue中的消息會被平攤給多個消費者。這時如果每個消息的處理時間不同，就有可能會導致某些消費者一直在忙，而另外一些消費者很快就處理完手頭工作並一直空閑的情況。我們可以通過設置prefetchCount來限制Queue每次發送給每個消費者的消息數，比如我們設置prefetchCount=1，則Queue每次給每個消費者發送一條消息；消費者處理完這條消息后Queue會再給該消費者發送一條消息
-
Exchange
在上一節我們看到生產者將消息投遞到Queue中，實際上這在RabbitMQ中這種事情永遠都不會發生。實際的情況是，生產者將消息發送到Exchange（交換器，下圖中的X），由Exchange將消息路由到一個或多個Queue中（或者丟棄）
-
routing key
生產者在將消息發送給Exchange的時候，一般會指定一個routing key，來指定這個消息的路由規則，而這個routing key需要與Exchange Type及binding key聯合使用才能最終生效。
在Exchange Type與binding key固定的情況下（在正常使用時一般這些內容都是固定配置好的），我們的生產者就可以在發送消息給Exchange時，通過指定routing key來決定消息流向哪里。
RabbitMQ為routing key設定的長度限制為255 bytes。 -
Binding
RabbitMQ中通過Binding將Exchange與Queue關聯起來，這樣RabbitMQ就知道如何正確地將消息路由到指定的Queue了。
Binding key 在綁定（Binding）Exchange與Queue的同時，一般會指定一個binding key；消費者將消息發送給Exchange時，一般會指定一個routing key；當binding key與routing key相匹配時，消息將會被路由到對應的Queue中。這個將在Exchange Types章節會列舉實際的例子加以說明。 在綁定多個Queue到同一個Exchange的時候，這些Binding允許使用相同的binding key。 binding key 並不是在所有情況下都生效，它依賴於Exchange Type，比如fanout類型的Exchange就會無視binding key，而是將消息路由到所有綁定到該Exchange的Queue。
二、生產者模型實例
-
生產者----隊列-----消費者 模型實例
(1)生產者代碼
require_once __DIR__ . '/vendor/autoload.php'; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; //創建連接 $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest'); //創建信道 $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();(2)消費者代碼
require_once __DIR__ . '/vendor/autoload.php'; use PhpAmqpLib\Connection\AMQPStreamConnection; //創建連接 $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest'); //創建信道 $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(); } $channel->close(); $connection->close();
2.生產者---交換機(fanout)---隊列-----消費者模型
交換機Fanout 類型，則會將消息發送給所有與該 Exchange 定義過 Binding 的所有 Queues 中去，其實是一種廣播行為。
(1)生產者代碼
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
//創建連接
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
//創建信道
$channel = $connection->channel();
//聲明fanout類型交換機
$channel->exchange_declare('logs', 'fanout', false, false, false);
$data = implode(' ', array_slice($argv, 1));
if (empty($data)) {
$data = "info: Hello World!";
}
$msg = new AMQPMessage($data);
//發送消息給交換機
$channel->basic_publish($msg, 'logs');
echo ' [x] Sent ', $data, "\n";
$channel->close();
$connection->close();
(2)消費者代碼
<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
//創建連接
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
//創建信道
$channel = $connection->channel();
//聲明交換機
$channel->exchange_declare('logs', 'fanout', false, false, false);
//臨時隊列,queue名字是空字符串的隊列
list($queue_name, ,) = $channel->queue_declare("", false, false, true, false);
//把隊列和交換機綁定
$channel->queue_bind($queue_name, 'logs');
echo " [*] Waiting for logs. To exit press CTRL+C\n";
$callback = function ($msg) {
echo ' [x] ', $msg->body, "\n";
};
//消費者消費消息
$channel->basic_consume($queue_name, '', false, true, false, false, $callback);
//等待消息
while (count($channel->callbacks)) {
$channel->wait();
}
$channel->close();
$connection->close();
3.生產者---交換機(direct)---隊列-----消費者 模型
direct類型的Exchange路由規則也很簡單，它會把消息路由到那些binding key與routing key完全匹配的Queue中
以上圖的配置為例，我們以routingKey=”error”發送消息到Exchange，則消息會路由到Queue1（amqp.gen-S9b…，這是由RabbitMQ自動生成的Queue名稱）和Queue2（amqp.gen-Agl…）；如果我們以routingKey=”info”或routingKey=”warning”來發送消息，則消息只會路由到Queue2。如果我們以其他routingKey發送消息，則消息不會路由到這兩個Queue中。
//info 是routing key
(1)生產者代碼
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
//創建連接
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
//創建信道
$channel = $connection->channel();
//聲明direct類型交換機
$channel->exchange_declare('direct_logs', 'direct', false, false, false);
$severity = isset($argv[1]) && !empty($argv[1]) ? $argv[1] : 'info';
$data = implode(' ', array_slice($argv, 2));
if (empty($data)) {
$data = "Hello World!";
}
$msg = new AMQPMessage($data);
//綁定交換機和路由key
$channel->basic_publish($msg, 'direct_logs', $severity);
echo ' [x] Sent ', $severity, ':', $data, "\n";
$channel->close();
$connection->close();
(2)消費者代碼
php .\receive_logs_direct.php info
info是 bind key
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
//創建連接
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
//創建信道
$channel = $connection->channel();
//聲明direct類型交換機
$channel->exchange_declare('direct_logs', 'direct', false, false, false);
list($queue_name, ,) = $channel->queue_declare("", false, false, true, false);
$severities = array_slice($argv, 1);
if (empty($severities)) {
file_put_contents('php://stderr', "Usage: $argv[0] [info] [warning] [error]\n");
exit(1);
}
foreach ($severities as $severity) {
//綁定交換機/隊列/路由key(綁定key)
$channel->queue_bind($queue_name, 'direct_logs', $severity);
}
echo " [*] Waiting for logs. To exit press CTRL+C\n";
$callback = function ($msg) {
echo ' [x] ', $msg->delivery_info['routing_key'], ':', $msg->body, "\n";
};
$channel->basic_consume($queue_name, '', false, true, false, false, $callback);
while (count($channel->callbacks)) {
$channel->wait();
}
$channel->close();
$connection->close();
4.生產者---交換機(topic)---隊列-----消費者 模型
direct類型的Exchange路由規則也很簡單，它會把消息路由到那些binding key與routing key完全匹配的Queue中。
圖解:
以上圖的配置為例，我們以routingKey=”error”發送消息到Exchange，則消息會路由到Queue1（amqp.gen-S9b…，這是由RabbitMQ自動生成的Queue名稱）和Queue2（amqp.gen-Agl…）；如果我們以routingKey=”info”或routingKey=”warning”來發送消息，則消息只會路由到Queue2。如果我們以其他routingKey發送消息，則消息不會路由到這兩個Queue中。
前面講到direct類型的Exchange路由規則是完全匹配bindingkey與routingkey，但這種嚴格的匹配方式在很多情況下不能滿足實際業務需求。topic類型的Exchange在匹配規則上進行了擴展，它與direct類型的Exchage相似，也是將消息路由到binding key與routing key相匹配的Queue中，但這里的匹配規則有些不同，它約定：
- routing key為一個句點號“. ”分隔的字符串（我們將被句點號“.”分隔開的每一段獨立的字符串稱為一個單詞），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”
- binding key與routing key一樣也是句點號“. ”分隔的字符串
- binding key中可以存在兩種特殊字符“”與“#”，用於做模糊匹配，其中“”用於匹配一個單詞，“#”用於匹配多個單詞（可以是零個）
(1)生產者代碼
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
//php emit_log_topic.php a.a data111
//a.a routingkey
//data1 數據
//聲明鏈接
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
//聲明信道
$channel = $connection->channel();
//聲明topic 類型交換機
$channel->exchange_declare('topic_logs', 'topic', false, true, false);
$routing_key= 'a.a';
$routing_key = isset($argv[1]) && !empty($argv[1]) ? $argv[1] : 'anonymous.info';
$data = implode(' ', array_slice($argv, 2));
if (empty($data)) {
$data = "Hello World123!";
}
//delivery_mode 2: 消息持久化
$msg = new AMQPMessage(
$data,
array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT)
);
$channel->basic_publish($msg, 'topic_logs', $routing_key);
echo ' [x] Sent ', $routing_key, ':', $data, "\n";
$channel->close();
$connection->close();
(2)消費者代碼
//php receive_logs_topic.php two a.*
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
//創建連接
$connection = new AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest');
//創建信道
$channel = $connection->channel();
//聲明topic類型交換機
$channel->exchange_declare('topic_logs', 'topic', false, true, false);
list($queue_name) = array_slice($argv, 1,1);
if (empty($queue_name)) {
file_put_contents('php://stderr', "Usage: $argv[0] [queue_name]\n");
exit(1);
}
//第四個參數：消費者是否唯一
$channel->queue_declare($queue_name, false, true, false, false);
$binding_keys = array_slice($argv, 2);
if (empty($binding_keys)) {
file_put_contents('php://stderr', "Usage: $argv[0] [binding_key]\n");
exit(1);
}
foreach ($binding_keys as $binding_key) {
$channel->queue_bind($queue_name, 'topic_logs', $binding_key);
}
echo " [*] Waiting for logs. To exit press CTRL+C\n";
//在RabbitMQ中有一個prefetch_count的概念，這個參數的意思是允許Consumer最多同時處理幾個任務。我的版本的RabbitMQ默認這個參數是3，也就是說如果某一個Consumer在收到消息后沒有發送ACK確認包，RabbitMQ就會任務Consumer還在處理任務，當有3個消息都沒有發送ACK確認包時，RabbitMQ就不會再發送消息給該Consumer。
$callback = function ($msg) {
sleep(3);
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
echo ' [x] ', $msg->delivery_info['routing_key'], ':', $msg->body, "\n";
};
$channel->basic_consume($queue_name, '', false, false, false, false, $callback);
while (count($channel->callbacks)) {
$channel->wait();
}
$channel->close();
$connection->close();