路由(Routing)

在前面的教程中,我们实现了一个简单的日志系统。可以把日志消息广播给多个接收者。

本篇教程中我们打算新增一个功能——使得它能够只订阅消息的一个字集。例如,我们只需要把严重的错误日志信息写入日志文件(存储到磁盘),但同时仍然把所有的日志信息输出到控制台中

绑定(Bindings)

前面的例子,我们已经创建过绑定(bindings),代码如下:

$exchange->publish($message, '');

绑定(binding)是指交换器(exchange)和队列(queue)的关系。可以简单理解为:这个队列(queue)对这个交换器(exchange)的消息感兴趣。

绑定的时候可以带上一个额外的 routing_key 参数。为了避免与 basic_publish 的参数混淆,我们把它叫做 binding key。以下是如何创建一个带 binding key 的绑定。

$exchange->publish($message, $routeKey);

binding key 的含义取决于交换器(exchange)的类型。我们之前使用过的 fanout 类型会忽略这个值。

Direct 类型的交换器(exchange)

我们的日志系统广播所有的消息给所有的消费者(consumers)。我们打算扩展它,使其可以能够精确的过滤消息。例如我们也许值是希望当接收到一个严重的错误的时候才把消息写入磁盘,以免浪费磁盘空间。

我们使用的 fanout 类型的交换器(exchange)扩展性不够——它能做的仅仅是广播。

我们将会使用 direct 类型的交换器(exchange)来代替。路由的算法很简单——交换器将会对 binding key 和 routing key 进行精确匹配,从而确定消息该分发到哪个队列。

下图能够很好的描述这个场景:

在这个场景中,我们可以看到 direct exchange X 和两个队列绑定了。第一个队列使用 orange 作为 binding key,第二个队列有两个绑定,一个使用 black 作为 binding key,另外一个是 green。

这样以来,当 routing key 为 orange 的消息发布到交换器(exchange),就会被路由到队列 Q1。routing key 为 black 或者 green 的消息就会路由到 Q2。其他的所有消息都将会被丢弃。

多个绑定(Multiple bindings)

多个队列使用相同的 binding key 是合法的。我们的这个例子,我们可以添加一个 X 和 Q1 之间的绑定,使用 blackbinding key。这样一来,direct 交换器就和 fanout 交换器的行为一样,将会广播消息到所有匹配的队列。带有 routing key 为 black 的消息都会发送到 Q1 和 Q2。

Emitting logs

我们将会发送消息到一个 direct exchange,把日志级别作为 routing key。这样子负责处理接收的脚本就可以选择它要处理的日志级别。我们先看看触发日志。

我们需要创建一个交换器(exchange):

$exchange->setName('direct_logs');

然后我们发送一则消息:

$exchange->publish($message, $severity);

我们先假设“severity”的值是 info、warning、error 中的一个。

订阅(Subscribing)

处理接收消息的方式和之前差不多,但是我们为每一个日志级别创建了一个新的绑定。

foreach ($severities as $item) {
    $queue->bind($exchangeName, $item);
}

整合

emit_log_direct.py 的代码:

<?php

/**
 * PHP amqp(RabbitMQ) Demo-4
 * @author  yuansir <yuansir@live.cn/yuanxuxu.com>
 */
$severity = count($argv) > 2 ? $argv[1] : 'info';
$message = empty($argv[2]) ? 'Hello World!' : ' ' . $argv[2];

$connection = new AMQPConnection(array('host' => '127.0.0.1', 'port' => '5672', 'vhost' => '/', 'login' => 'guest', 'password' => 'guest'));
$connection->connect() or die("Cannot connect to the broker!\n");

$channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);
$exchange->setName('direct_logs');
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->declare();

$exchange->publish($message, $severity);
var_dump("[x] Sent $message");

$connection->disconnect();

receive_logs_direct.py 的代码:

<?php

/**
 * PHP amqp(RabbitMQ) Demo-4
 * @author  yuansir <yuansir@live.cn/yuanxuxu.com>
 */
$exchangeName = 'direct_logs';

$connection = new AMQPConnection(array('host' => '127.0.0.1', 'port' => '5672', 'vhost' => '/', 'login' => 'guest', 'password' => 'guest'));
$connection->connect() or die("Cannot connect to the broker!\n");
$channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->declare();
$queue = new AMQPQueue($channel);
$queue->setFlags(AMQP_EXCLUSIVE);
$queue->declare();

$severities = $argv;
$file = $severities[0];
unset($severities[0]);
if (!$severities) {
        var_dump("Usage:$file [info] [warning] [error]");
        exit();
} else {
        foreach ($severities as $item) {
                $queue->bind($exchangeName, $item);
        }
}

var_dump('[*] Waiting for messages. To exit press CTRL+C');
while (TRUE) {
        $queue->consume('callback');
}
$connection->disconnect();

function callback($envelope, $queue) {
        $msg = $envelope->getBody();
        var_dump('[x]' . $envelope->getRoutingKey() . ':' . $msg);
        $queue->nack($envelope->getDeliveryTag());
}

如果你希望只是保存 warning 和 error 级别的日志到磁盘,只需要打开控制台并输入:

$ php receive_logs_direct.php warning error &gt; logs_from_rabbit.log

如果你希望所有的日志信息都输出到屏幕中,打开一个新的终端,然后输入:

$ php receive_logs_direct.php info warning error
 [*] Waiting for logs. To exit press CTRL+C

如果要触发一个 error 级别的日志,只需要输入:

$ php emit_log_direct.php error "Run. Run. Or it will explode."
 [x] Sent 'error':'Run. Run. Or it will explode.'

转载请注明: 转载自Ryan 是菜鸟 | LNMP 技术栈笔记

如果觉得本篇文章对您十分有益,何不 打赏一下

谢谢打赏

本文链接地址: RabbitMQ 官方中文入门教程(PHP 版) 第四部分:路由(Routing)

知识共享许可协议 本作品采用知识共享署名-非商业性使用 4.0 国际许可协议进行许可