RabbitMQ 是最流行的开源消息中间件之一。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面都非常的优秀。
消息中间件主要用来解决应用的耦合、异步处理和流量削锋。

相关概念

Channel

消息通道，在客户端的每个连接里，可建立多个 channel ，每个 channel 代表一个会话任务。

$connection = new AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest');
$channel = $connection->channel();

Queue

消息队列最终存储的载体，每个消息都会被发送到一个或多个队列。
消息实际上不直接发送到队列，而是发送到 Exchange，再由 Exchange 决定将消息投放到一个或多个队列。
如果没有指定 Exchange，则会发送到 RabbitMQ 默认的 Exchange 中（RabbitMQ 安装后会带有不同类型的 Exchange ）。
定义队列时，如果没有指定队列名称，会自动生成队列名称。

<?php
list($queue_name, ,) = $channel->queue_declare("", false, false, true, false);

Exchange

消息交换机。消息生产者将消息发送到 exchange，由 exchange 按对应规则将消息路由到不同消息队列。
发送消息过程一般是先定义消息 exchange ，然后发送消息到 exchange, 如下代码：

$channel->exchange_declare('exchange_name', $exchange_type, false, false, false);
$channel->basic_publish($msg, $exchange, $routing_key);

不同的 exchange type，消息路由到队列的规则不同。

fanout 类型

exchange 不会管 routing_key ，它会把消息广播到每个和它连接的 queue 中。

direct 类型

exchange 会检测消息中的 routing_key ，然后发送到用相同routing_key 进行绑定的 queue 中。
如下图，如果消息中的routing_key 为 info，则只发到 gen-Ag1，如果是 error ，则发到与之绑定的两个队列。

direct.png

topic 类型

它是 direct 类型的进阶版。它支持通配符的匹配，可以通过一个 routing_key 将消息发送到多个 queue。故通常 routing_key 会是一个包含“.”分割多个单词。

topic.png

”*”匹配单个单词，”#”匹配多个单词。

headers 类型

exchange 通过监测消息的头信息，而不是 routing key 来决定消息的发送。

exchange 其他属性

持久化
自动删除（没有队列绑定时是否自动删除）

Binding

绑定，它的作用是定义 exchange 发送消息到 queue 的规则。

$channel->queue_bind($queue_name, $exchange_name, $routing_key);

绑定时的 routing key 与过滤器相似。

Routing Key

路由关键字，exchange 根据这个关键字进行消息的发送。

Vhost

虚拟主机，一个 RabbitMQ 服务里可以创建多个 vhost ，用作不同用户的权限分离。

示例

发送消息

<?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();

其他

1. 在消费消息时，可以将主机名和进程号作为消费者标记，这样方便后面消费异常时查询原因。
2. 如果生产者没有定义 exchange，则实际是发送到 RabbitMQ 默认的 exchange 中（包含4种不同 exchange type 类型的 exchange ），并且默认以发送的队列名称为对应的 routing key。
3. 不可以定义名称相同，而又属性不同的消息队列。
4. 如果要让消息持久化，不仅要定义持久化消息队列，还要对消息标记持久化。