RabbitMQ 入门 - 话题

话题

本文基于 官方文档 翻译

(using php-amqplib)

在之前的教程中，我们改进了日志记录系统。 我们没有使用只有虚拟广播的 fanout （扇出）交换器，而是使用了直接交换器，并获得了选择性接收日志的可能性。

尽管使用直接交换改进了我们的系统，但它仍然有局限性 - 它不能根据多个标准进行路由。

在我们的日志系统中，我们可能不仅需要根据严重性来订阅日志，还要根据发布日志的来源进行订阅。 您可能从 syslog unix 工具知道这个概念，该工具根据严重性（info/warn/crit...）和工具（auth/cron/kern...）来路由日志。

这会给我们很大的灵活性 - 我们可能想听取来自 'cron' 的严重错误，而且还听取来自 'kern' 的所有日志。

为了在我们的日志系统中实现这一点，我们需要了解更复杂的 topicexchange。

话题交换

发送到话题交换的消息不能有任意的 routing_key - 它必须是由点分隔的单词列表。 单词可以是任何东西，但通常它们指定了与该消息相关的一些功能。

一些有效的路由键例子：“stock.usd.nyse”，“nyse.vmw”，“quick.orange.rabbit”。 只要您愿意，路由键中可以有多少个字，最多255个字节。

绑定键也必须是相同的形式。 话题交换背后的逻辑类似于直接话题 - 使用特定路由键发送的消息将被传递到与匹配绑定键绑定的所有队列。 但是绑定键有两个重要的特殊情况：

• * (star) 可以替代一个字。
• # (hash) 可以替代零个或更多的单词。

用一个很简单例子可以解释：

在这个例子中，我们将发送所有描述动物的消息。 消息将使用由三个字（两个点）组成的路由键发送。 路由关键字中的第一个单词将描述速度，第二个颜色和第三个种类：“ .. ”。

我们创建了三个绑定：Q1 绑定了 “ .orange. ” 和 Q2 绑定了 “ ..rabbit ” 和 “ lazy.# ” 。

These bindings can be summarised as:

这些绑定可以概括为：

• Q1 对所有的橙色动物（*.orange.*）都感兴趣。
• Q2 希望监听关于兔子（*.*.rabbit）的一切，以及关于懒惰动物（lazy.#）的一切。

将路由键设置为 “quick.orange.rabbit” 或 “lazy.orange.elephant” ，它们的消息都将传递到两个队列。 另一方面，“quick.orange.fox” 只会进入第一个队列，而 “lazy.brown.fox” 只会进入第二个队列。 “lazy.pink.rabbit” 只会传递到第二个队列一次，即使它匹配了两个绑定。 “quick.brown.fox” 不匹配任何绑定，因此将被丢弃。

如果我们违反我们的契约并发送带有一个或四个单词的消息，如 “orange” 或 “quick.orange.male.rabbit” ，会发生什么情况？ 答案是，这些消息将不匹配任何绑定并且丢失。

另一方面，“lazy.orange.male.rabbit” 即使有四个单词，也会匹配最后一个绑定，并将传递到第二个队列。

话题交换

话题交换功能强大，可以像其他交流一样行事。

当使用 “＃”(hash) 绑定键绑定队列时，它将接收所有消息，而不管路由密钥如何 - 就像在 fanout （扇出）交换中一样。

当在绑定中没有使用特殊字符 “*”（星号）和 “＃”（hash）时，主题交换将像直接交换一样。

把它们放在一起

我们将在我们的日志系统中使用主题交换。 我们首先假设日志的路由键有两个字：“.” 。

代码几乎与前一个教程中的代码相同。

emit_log_topic.php 的代码：

<?php

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->exchange_declare('topic_logs', 'topic', false, false, false);

$routing_key = isset($argv[1]) && !empty($argv[1]) ? $argv[1] : 'anonymous.info';
$data = implode(' ', array_slice($argv, 2));
if(empty($data)) $data = "Hello World!";

$msg = new AMQPMessage($data);

$channel->basic_publish($msg, 'topic_logs', $routing_key);

echo " [x] Sent ",$routing_key,':',$data," \n";

$channel->close();
$connection->close();

?>

receive_logs_topic.php 的代码：

<?php

require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

$connection = new AMQPStreamConnection('localhost', 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_slice($argv, 1);
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";

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

?>

接收所有日志：

php receive_logs_topic.php "#"

只从接收 “kern” 话题日志：

php receive_logs_topic.php "kern.*"

或者，如果您只想接收关于 “关键” 的日志信息：

php receive_logs_topic.php "*.critical"

您可以创建多个绑定：

php receive_logs_topic.php "kern.*" "*.critical"

并发布带有路由键 “kern.critical” 类型的日志：

php emit_log_topic.php "kern.critical" "A critical kernel error"

玩这些程序玩得开心。 请注意，代码没有对路由或绑定键作任何假设，您可能需要使用两个以上的路由键参数。

(Full source code for emit_log_topic.php and receive_logs_topic.php)

接下来，在教程6中，了解如何将远程过程调用作为远程过程调用

本作品采用《CC 协议》，转载必须注明作者和本文链接