RabbitMQ 路由模式(Routing)

有选择地接收消息

在前面的教程中我们建立了一个简单的日志记录系统。我们可以播放许多接收器日志消息。

在本教程中,我们将添加一个特性——我们将可以订阅信息的一个子集。例如,我们将能够直接只关键错误消息日志文件(为了节省磁盘空间),同时仍然能够在控制台上打印所有的日志消息。

绑定

在之前的例子我们已经创建绑定。你可能记得代码:

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

绑定是关联了exchange和quque。这可以简单地理解为:queue只感兴趣这个的exchange。

绑定可以额外routing_key参数,为了避免混乱$channel::basic_publish参数我们称之为 binding _key,这是我们如何创建一个binding_key:

  1. $binding_key = 'black';
  2. $channel->queue_bind($queue_name, $exchange_name, $binding_key);

binding_key依赖于exchange type。fanout exchanges,我们使用之前,这个参数是被忽略的。

Direct exchange

我们的日志系统从前面的教程广播所有消费者的所有消息。我们想扩展允许过滤消息根据其严重性。例如,我们可能希望日志消息写入磁盘的脚本只接收关键错误,而不是浪费磁盘空间警告或信息日志消息。

我们使用一个fanout exchanges,不会给我们太多的灵活性——这只是盲目的广播的能力。

我们将使用一个direct exchange。direct exchange背后的路由算法很简单——一个消息的队列binding_key完全匹配message的routing_key。

为了说明这一点,考虑下面的设置

img

在上图中: exchange X和两个queue绑定在一起。queue Q1的binding key是orange。queue Q2的binding key是black和green。 当P publish key是orange时,exchange会把它放到Q1。如果是black或者green那么就会到Q2。其余的Message都会被丢弃。

Multiple bindings

img

多个queue绑定到相同的binding_key是合法的。在我们的例子中,我们可以添加一个绑定X和Q1之间绑定关键黑色。在这种情况下,直接交换将像扇出,广播到所有匹配的消息队列。消息路由关键黑色将Q1和Q2。

多个queue绑定到相同的binding_key是合法的。这个的例子中,Q1和Q2都绑定了black。也就是说,对于routing_key是black的Message,会被放到到Q1和Q2。其余的Message都会被丢弃。这样的direct exchange很像fanout exchange

发送日志

我们将使用这个模型我们的日志系统。而不是扇出我们会发送消息到一个直接交换。我们将提供日志严重性作为路由的关键。这样接收脚本将能够选择严重性要收到。让我们先关注排放日志。

像往常一样,我们首先需要创建一个交流:

  1. $channel->exchange_declare('direct_logs', 'direct', false, false, false);

我们准备发送消息:

  1. $channel->exchange_declare('direct_logs', 'direct', false, false, false);
  2. $channel->basic_publish($msg, 'direct_logs', $severity);

简化我们假定“严重性”可以是一个‘信息’,‘警告’,‘错误’。

订阅

接收消息的工作就像在前面的教程中,有一个例外,我们要创建一个新绑定每个严重性我们感兴趣。

  1. foreach($severities as $severity) {
  2. $channel->queue_bind($queue_name, 'direct_logs', $severity);
  3. }

把它放在一起

img

emit_log_direct.php代码类:

  1. <?php
  2. require_once __DIR__ . '/vendor/autoload.php';
  3. use PhpAmqpLib\Connection\AMQPStreamConnection;
  4. use PhpAmqpLib\Message\AMQPMessage;
  5. $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
  6. $channel = $connection->channel();
  7. $channel->exchange_declare('direct_logs', 'direct', false, false, false);
  8. $severity = isset($argv[1]) && !empty($argv[1]) ? $argv[1] : 'info';
  9. $data = implode(' ', array_slice($argv, 2));
  10. if(empty($data)) $data = "Hello World!";
  11. $msg = new AMQPMessage($data);
  12. $channel->basic_publish($msg, 'direct_logs', $severity);
  13. echo " [x] Sent ",$severity,':',$data," \n";
  14. $channel->close();
  15. $connection->close();

receive_logs_direct.php代码:

  1. <?php
  2. require_once __DIR__ . '/vendor/autoload.php';
  3. use PhpAmqpLib\Connection\AMQPStreamConnection;
  4. $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
  5. $channel = $connection->channel();
  6. $channel->exchange_declare('direct_logs', 'direct', false, false, false);
  7. list($queue_name, ,) = $channel->queue_declare("", false, false, true, false);
  8. $severities = array_slice($argv, 1);
  9. if(empty($severities )) {
  10. file_put_contents('php://stderr', "Usage: $argv[0] [info] [warning] [error]\n");
  11. exit(1);
  12. }
  13. foreach($severities as $severity) {
  14. $channel->queue_bind($queue_name, 'direct_logs', $severity);
  15. }
  16. echo ' [*] Waiting for logs. To exit press CTRL+C', "\n";
  17. $callback = function($msg){
  18. echo ' [x] ',$msg->delivery_info['routing_key'], ':', $msg->body, "\n";
  19. };
  20. $channel->basic_consume($queue_name, '', false, true, false, false, $callback);
  21. while(count($channel->callbacks)) {
  22. $channel->wait();
  23. }
  24. $channel->close();
  25. $connection->close();

如果你想节省只有“警告”和“错误”(而不是“信息”)日志消息到一个文件,打开一个控制台和类型:

  1. php receive_logs_direct.php warning error > logs_from_rabbit.log

如果你想看到所有的日志消息在你的屏幕上,打开一个新的终端和做的事:

  1. php receive_logs_direct.php info warning error
  2. # => [*] Waiting for logs. To exit press CTRL+C

例如,发出一个错误日志消息类型:

  1. php emit_log_direct.php error "Run. Run. Or it will explode."
  2. # => [x] Sent 'error':'Run. Run. Or it will explode.'