RabbitMQ 入门指南（三）

分手后的思念是犯贱 2022-11-21 01:11 175阅读 0赞

## Publish/Subscribe ##

**发布/订阅模式，可以将一条消息同时发送给多个 consumer。**

为了举例说明该模式，我们将构建一个简单的日志系统。它包含两个程序，第一个程序用于发布日志消息，第二个用于接收并打印日志消息。

每一个运行中的 receiver 程序副本都将接收消息。一个 receiver 用来将日志存储到磁盘；同时，另一个 receiver 用来查看日志。

本质上，已发布的日志 message 将被广播给所有的 receiver。

![image][]

## Exchange ##

RabbitMQ 中的消息模型的核心思想是 producer 不会直接把消息发送给队列。事实上，多数情况下 producer 根本就不知道消息是否会被发送给队列。

取而代之的是，producer 只是把消息发送给 **exchange（交换机）**。

**exchange 负责把来自生产者的消息，按照一定的路由规则，发送给与其建立了绑定关系的队列。**

exchange 一边接收来自 producer 的消息，一边将消息推送到队列。

exchange 必须准确地知道如何处理来自 producer 的消息：

*  消息是否该被推送到一个指定的队列？
 *  消息是否该被推送到多个队列？
 *  消息是否该被丢弃？

这些规则都是由 exchange type 定义的。

### exchange type ###

exchange type 主要包含以下四种：

*  direct： 它会把消息路由给 binding\_key 与 routing\_key 完全匹配的队列。
 *  topic： 与 direct 交换机类似，也是将消息路由给 binding\_key 与 routing\_key 相匹配的队列。 只不过，topic 交换机不是完全匹配，它的匹配规则更加灵活，支持模糊匹配。
 *  headers： 它会把消息路由给 binding\_key（键值对）与消息的 headers 中的键值对完全匹配的队列。
 *  fanout： 顾名思义，它会把收到的所有消息广播给所有与它绑定的队列。

这里，我们使用 fanout，因为它刚好是日志收集系统所需要的 exchange type。

### fanout ###

创建名称为 logs 的 fanout 类型的交换机。

$channel->exchange_declare('logs', 'fanout', false, false, false);

fanout exchange 非常简单，顾名思义，它会把收到的所有消息广播给所有与它绑定的队列。

### 列出所有的交换机 ###

# 列出服务器中所有的交换机
    rabbitmqctl list_exchanges

输出结果：

Listing exchanges for vhost / ...
    name                    type
    amq.rabbitmq.trace      topic
    amq.match               headers
    amq.direct              direct
    amq.fanout              fanout
                            direct
    amq.topic               topic
    amq.headers             headers

上面这些是 RabbitMQ 中默认就已经创建好的 exchange。此刻，你不太可能都要去使用它们。

### 默认交换机 ###

上面，我们发现有个**交换机的 name 是空字符串（没有名称），type 是 direct，它就是默认的交换机。**

回顾之前的入门指南，我们根本就不知道 exchange，却仍然可以发送消息到队列，是因为我们使用的是默认交换机。

之前，我们是这样发送消息给队列的。

// 将消息发送到指定的队列
    $channel->basic_publish($msg, '', 'task_queue');

basic\_publish() 方法的第二个参数就是 exchange。

上面，我们使用了默认的没有名称的交换机，消息会被路由给由 **routing\_key** 参数指定的队列。

如果 routing\_key 存在，routing\_key 是 basic\_publish() 方法的第三个参数。

现在，我们使用我们自己命名的交换机来发送消息。

$channel->exchange_declare('logs', 'fanout', false, false, false);
    $channel->basic_publish($msg, 'logs');

### routing\_key ###

routing\_key 是 basic\_publish() 方法的第三个参数。

当生产者发送消息给交换机时，一般会指定 routing\_key，来指定该消息的路由规则。

routing\_key 需要配合交换机的 type 和 binding\_key 一起使用才会生效。

### Temporary queue ###

先前，我们使用的队列都有具体的队列名称。当你想要在 producer 和 consumer 之间共享队列时，指定队列名称是非常重要的。

但是，它并不适用于我们现在的日志系统。我们想要得知所有的日志消息，而不仅是它的子集。而且，我们只关注当前正在流动的消息，而不是旧的消息。

**为了实现该目的，我们需要做两件事。**

*  首先，无论何时连接到 RabbitMQ，都需要一个崭新的空队列。为此，我们可以创建一个随机名称的队列，或者更好，由服务器来提供一个随机的队列名称。
 *  其次，一旦关闭 consumer 的连接，队列应该自动被删除。

对于 php-amqplib 客户端，当我们提供一个空字符串的 queue\_name，就可创建一个非持久的带有已生成名称的队列。

list($queue_name, ,) = $channel->queue_declare("");

上例中，$queue\_name 变量的值是一个随机的、由 RabbitMQ 服务器生成的 queue name。例如，它的值可能为 `amq.gen-JzTY20BRgKO-HjmUJj0wLg`。

当连接声明关闭时，队列将被自动删除，因为该队列被声明为 exclusive（独有的、排他的）。

**这就是所说的 Temporary queue（临时队列）。**

如果你想要了解更多关于 exclusive 标签的信息或者其他的队列属性，请参考 [guide on queues][] 。

### Binding ###

上面，我们已经创建好了 fanout 类型的 exchange，现在需要**建立 exchange 和 queue 之间的关系，这就叫做 Binding（绑定）**。

$channel->queue_bind($queue_name, 'logs');

现在，logs 交换机就可以推送消息到我们绑定的队列了。

列出已存在的绑定

# 列出服务器中已存在的绑定
    rabbitmqctl list_bindings

## 最终的代码示例 ##

emit\_log.php，作为生产者，内容如下：

<?php
    
    // php 的依赖包自动加载，引入相关的类
    require_once __DIR__ . '/vendor/autoload.php';
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    use PhpAmqpLib\Message\AMQPMessage;
    
    // 连接到指定的 RabbitMQ 服务器（Broker），并创建 channel
    $connection = new AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest');
    $channel = $connection->channel();
    
    // 声明 exchange
    $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);
    
    // 将消息发送到指定的 exchange，再路由到与其绑定的 queue
    $channel->basic_publish($msg, 'logs');
    
    // 消息发送成功后，输出提示信息
    echo ' [x] Sent ', $data, "\n";
    
    // 最后，关闭 channel 和 connection
    $channel->close();
    $connection->close();

**说明：** 如果队列还没有绑定到交换机，先发送的消息就会丢失，但这并无大碍。如果消费者还没监听，交换机会自动丢弃消息。

receive\_logs.php，作为消费者，内容如下：

<?php
    
    // php 的依赖包自动加载，引入相关的类
    require_once __DIR__ . '/vendor/autoload.php';
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    
    // 连接到指定的 RabbitMQ 服务器（Broker），并创建 channel
    $connection = new AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest');
    $channel = $connection->channel();
    
    // 声明 exchange
    $channel->exchange_declare('logs', 'fanout', false, false, false);
    
    // 声明要监听的队列（queue）
    list($queue_name, ,) = $channel->queue_declare("", false, false, true, false);
    
    // 将队列绑定到 exchange
    $channel->queue_bind($queue_name, 'logs');
    
    // 提示信息
    echo " [*] Waiting for messages. To exit press CTRL+C\n";
    
    // 回调函数
    $callback = function ($msg) {
        // todo 处理消息的业务逻辑
        echo ' [x]', $msg->body, "\n";
    };
    
    // 启动消费者
    $channel->basic_consume($queue_name, '', false, true, false, false, $callback);
    
    while ($channel->is_consuming()) { // 如果正在进行回调处理
        $channel->wait(); // 等待
    }
    
    // 最后，关闭 channel 和 connection
    $channel->close();
    $connection->close();

运行两个消费者 C1 和 C2。

# C1 用于保存日志到磁盘文件
    php receive_logs.php > logs_from_rabbit.log

# C2 用于直接查看日志
    php receive_logs.php

运行一个生产者 P。

# P 用于产生和发送日志消息
    php emit_log.php

使用命令 `rabbitmqctl list_bindings` 可以验证我们的代码是否成功创建了绑定关系。

下一节，我们将介绍如何监听消息的子集。

**参考文献**

\[1\] [https://www.rabbitmq.com/tutorials/tutorial-three-php.html][https_www.rabbitmq.com_tutorials_tutorial-three-php.html]

[image]: /images/20221120/b6ec0225c3eb4f4e8d31428919772ddb.png
[guide on queues]: https://www.rabbitmq.com/queues.html
[https_www.rabbitmq.com_tutorials_tutorial-three-php.html]: https://www.rabbitmq.com/tutorials/tutorial-three-php.html