RabbitMQ 是最流行的開(kāi)源消息中間件之一永毅。用于在分布式系統(tǒng)中存儲(chǔ)轉(zhuǎn)發(fā)消息筑悴，在易用性、擴(kuò)展性癞季、高可用性等方面都非常的優(yōu)秀劫瞳。
消息中間件主要用來(lái)解決應(yīng)用的耦合、異步處理和流量削鋒绷柒。
相關(guān)概念
Channel
消息通道志于，在客戶端的每個(gè)連接里，可建立多個(gè) channel 废睦，每個(gè) channel 代表一個(gè)會(huì)話任務(wù)伺绽。
$connection = new AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest');
$channel = $connection->channel();
Queue
消息隊(duì)列最終存儲(chǔ)的載體，每個(gè)消息都會(huì)被發(fā)送到一個(gè)或多個(gè)隊(duì)列。
消息實(shí)際上不直接發(fā)送到隊(duì)列憔恳，而是發(fā)送到 Exchange瓤荔，再由 Exchange 決定將消息投放到一個(gè)或多個(gè)隊(duì)列。
如果沒(méi)有指定 Exchange钥组，則會(huì)發(fā)送到 RabbitMQ 默認(rèn)的 Exchange 中（RabbitMQ 安裝后會(huì)帶有不同類型的 Exchange ）输硝。
定義隊(duì)列時(shí)，如果沒(méi)有指定隊(duì)列名稱程梦，會(huì)自動(dòng)生成隊(duì)列名稱点把。
<?php
list($queue_name, ,) = $channel->queue_declare("", false, false, true, false);
Exchange
消息交換機(jī)。消息生產(chǎn)者將消息發(fā)送到 exchange屿附，由 exchange 按對(duì)應(yīng)規(guī)則將消息路由到不同消息隊(duì)列郎逃。
發(fā)送消息過(guò)程一般是先定義消息 exchange ，然后發(fā)送消息到 exchange, 如下代碼：
$channel->exchange_declare('exchange_name', $exchange_type, false, false, false);
$channel->basic_publish($msg, $exchange, $routing_key);
不同的 exchange type挺份，消息路由到隊(duì)列的規(guī)則不同褒翰。
fanout 類型
exchange 不會(huì)管 routing_key ，它會(huì)把消息廣播到每個(gè)和它連接的 queue 中匀泊。
direct 類型
exchange 會(huì)檢測(cè)消息中的 routing_key 优训，然后發(fā)送到用相同routing_key 進(jìn)行綁定的 queue 中。
如下圖各聘，如果消息中的routing_key 為 info揣非，則只發(fā)到 gen-Ag1，如果是 error 躲因，則發(fā)到與之綁定的兩個(gè)隊(duì)列早敬。
topic 類型
它是 direct 類型的進(jìn)階版。它支持通配符的匹配大脉，可以通過(guò)一個(gè) routing_key 將消息發(fā)送到多個(gè) queue搞监。故通常 routing_key 會(huì)是一個(gè)包含“.”分割多個(gè)單詞。
”*”匹配單個(gè)單詞镰矿，”#”匹配多個(gè)單詞腺逛。
headers 類型
exchange 通過(guò)監(jiān)測(cè)消息的頭信息，而不是 routing key 來(lái)決定消息的發(fā)送衡怀。
exchange 其他屬性
持久化
自動(dòng)刪除（沒(méi)有隊(duì)列綁定時(shí)是否自動(dòng)刪除）
Binding
綁定，它的作用是定義 exchange 發(fā)送消息到 queue 的規(guī)則安疗。
$channel->queue_bind($queue_name, $exchange_name, $routing_key);
綁定時(shí)的 routing key 與過(guò)濾器相似抛杨。
Routing Key
路由關(guān)鍵字，exchange 根據(jù)這個(gè)關(guān)鍵字進(jìn)行消息的發(fā)送荐类。
Vhost
虛擬主機(jī)怖现，一個(gè) RabbitMQ 服務(wù)里可以創(chuàng)建多個(gè) vhost ，用作不同用戶的權(quán)限分離。
示例
發(fā)送消息
<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
$connection = new AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->exchange_declare('topic_logs', 'topic', false, false, false);
$data = "Hello World!";
$msg = new AMQPMessage($data);
$routing_key = 'app.error';
$channel->basic_publish($msg, 'topic_logs', $routing_key);
echo " [x] Sent ".$routing_key.' : '. $data. "\n";
$channel->close();
$connection->close();
接受消息
<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
$connection = new AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->exchange_declare('topic_logs', 'topic', false, false, false);
list($queue_name, , ) = $channel->queue_declare("", false, false, true, false);
$binding_keys = array(
'app.error',
'h5.error',
);
foreach ($binding_keys as $binding_key) {
$channel->queue_bind($queue_name, 'topic_logs', $binding_key);
}
$callback = function ($msg) {
echo " [x] ". $msg->delivery_info['routing_key'].":".$msg->body."\n";
};
$consumer = gethostbyname().':'.getmypid();
$channel->basic_consume($queue_name, $consumer, false, true, false, false, $callback);
while (count($channel->callbacks)) {
$channel->wait();
}
$channel->close();
$connection->close();
其他
- 在消費(fèi)消息時(shí)屈嗤，可以將主機(jī)名和進(jìn)程號(hào)作為消費(fèi)者標(biāo)記潘拨，這樣方便后面消費(fèi)異常時(shí)查詢?cè)颉?/li>
- 如果生產(chǎn)者沒(méi)有定義 exchange，則實(shí)際是發(fā)送到 RabbitMQ 默認(rèn)的 exchange 中（包含4種不同 exchange type 類型的 exchange ）饶号，并且默認(rèn)以發(fā)送的隊(duì)列名稱為對(duì)應(yīng)的 routing key铁追。
- 不可以定義名稱相同，而又屬性不同的消息隊(duì)列茫船。
- 如果要讓消息持久化琅束，不僅要定義持久化消息隊(duì)列，還要對(duì)消息標(biāo)記持久化算谈。
