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 ，则发到与之绑定的两个队列。
topic 类型
它是 direct 类型的进阶版。它支持通配符的匹配，可以通过一个 routing_key 将消息发送到多个 queue。故通常 routing_key 会是一个包含“.”分割多个单词。
”*”匹配单个单词，”#”匹配多个单词。
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();
其他
- 在消费消息时，可以将主机名和进程号作为消费者标记，这样方便后面消费异常时查询原因。
- 如果生产者没有定义 exchange，则实际是发送到 RabbitMQ 默认的 exchange 中（包含4种不同 exchange type 类型的 exchange ），并且默认以发送的队列名称为对应的 routing key。
- 不可以定义名称相同，而又属性不同的消息队列。
- 如果要让消息持久化，不仅要定义持久化消息队列，还要对消息标记持久化。
