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

路由(Routing)

在前面的教程中，咱们实现了一个简单的日志系统。能够把日志消息广播给多个接收者。

本篇教程中咱们打算新增一个功能——使得它可以只订阅消息的一个字集。例如，咱们只须要把严重的错误日志信息写入日志文件（存储到磁盘），但同时仍然把全部的日志信息输出到控制台中

绑定（Bindings）

前面的例子，咱们已经建立过绑定（bindings），代码以下：

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

绑定（binding）是指交换器（exchange）和队列（queue）的关系。能够简单理解为：这个队列（queue）对这个交换器（exchange）的消息感兴趣。

绑定的时候能够带上一个额外的routingkey参数。为了不与basicpublish的参数混淆，咱们把它叫作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);
}

整合

emitlogdirect.py的代码：

<?php

/**
 * PHP amqp(RabbitMQ) Demo-4
 * @author  yuansir &amp;lt;yuansir@live.cn/yuansir-web.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();

receivelogsdirect.py的代码：

<?php

/**
 * PHP amqp(RabbitMQ) Demo-4
 * @author  yuansir &amp;lt;yuansir@live.cn/yuansir-web.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 > 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.'