RabbitMq学习5-路由(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进行精确匹配,
从而确定消息该分发到哪个队列。
下图能够很好的描述这个场景:
四、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);
}
六、整合
1、生产者:emitlogdirect.php
?php /**
* PHP amqp(RabbitMQ) Demo-4
*/
$severity = count($argv) > 2 ? $argv[1] : 'info';
$message = empty($argv[2]) ? 'Hello World!' : ' ' . $argv[2]; var_dump($argv); $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->declareExchange(); $exchange->publish($message, $severity);//发送消息 var_dump("[x] Sent $message"); $connection->disconnect();
2、消费者:receivelogsdirect.php
<?php /**
* PHP amqp(RabbitMQ) Demo-4
* @author yuansir <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->declareExchange(); $queue = new AMQPQueue($channel);
$queue->setFlags(AMQP_EXCLUSIVE);
$queue->declareQueue(); $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 receivelogsdirect.php warning error > logs_from_rabbit.log
如果你希望所有的日志信息都输出到屏幕中,打开一个新的终端,然后输入:
$ php receivelogsdirect.php info warning error
[*] Waiting for logs. To exit press CTRL+C
如果要触发一个error级别的日志,只需要输入:
$ php emitlogdirect.php error "Run. Run. Or it will explode."
[x] Sent 'error':'Run. Run. Or it will explode.'
最新文章
- 易图软件之EaseMap Desktop 1.0发布
- 自定义响应结构 Json格式转换 工具类
- Redis(一) 介绍
- 入门React和Webpack
- 【转】webGL与OpenGL的不同
- C++语言-05-三大特性
- iphone删除自动更新的系统
- Python 打开目录与指定文件
- 连续值的CART(分类回归树)原理和实现
- dpi,ppi,dip,dp,px和sp
- Java中的private、protected、public和default的区别
- [转]Nuget挂了的解决方法
- 通过shell脚本实现代码自动化部署
- 用Javascript的for循环输出质数
- MySQL filesort优化案例一则
- Docker教程:使用docker配置python开发环境
- linux基础07-bash编程(变量,变量类型)
- leetcode937-java
- C#编程(四十)----------运算符重载
- Chapter 3 -- Ordering