RabbitMQ tutorial 5 - Topics

I continue the series of translation lessons from the official site . Examples will be on php, but they can be implemented on most popular PL .

In the previous article, we improved the logging system. Instead of a fanout access point (which is only suitable for elementary message translation), we used direct - and were able to receive messages through certain samples.

Although direct has perfected our system, it still has a flaw - it cannot compose a route (routing) according to several criteria.
')
For example, we needed to separate the logs not only by its type of importance, but also by the source of the log. You have probably encountered such a concept in the unix syslog tool, which is distinguished by logs by its type of importance (info / warn / crit ...) and by its object (auth / cron / kern ...).

We gain flexibility in the request. For example, we can get all the logs with the error type that came from 'cron'-a, and all the logs that came from' kern '. In order to implement this in our logging system, let's examine the access point - topic.

Topic

Messages sent to the topic access point cannot be sent with an arbitrary routing_key key - it must be a list of words separated by a point. Words can be any, but usually they are associated with some properties of the message. Here are examples of proper routing_key: “stock.usd.nyse”, “nyse.vmw”, “quick.orange.rabbit”. The key length should not exceed 255 bytes.

Binding key is made by the same rule. The logic of the topic is the same as for direct - messages reach the queues, the binding key of which coincides with the routing key of the message. But there are 2 special features for the topic:

• * (star) can be replaced by exactly 1 word;
• # (hash) can be replaced with 0 or more words.

It will be clearer to show in the figure:



In this example, we send letters to all given animals. Messages contain a routing key consisting of 3 words (with two dots). The first word describes speed, the second color and the third type: "speed.colour.species".

The Q1 queue will be associated with the "* .orange. *" Key, and the Q2 queue will be connected with the "*. *. Rabbit" and "lazy. #" Keys.

Links can be summarized as:

1. The Q1 queue is interested in all orange animals;
2. The Q2 line wants to know everything about rabbits and lazy animals.

The message with the key “quick.orange.rabbit” will reach both queues. The message with the key "lazy.orange.elephant" will also come in both queues. A message with the “quick.orange.fox” type will only reach 1 queue, and with the “lazy.brown.fox” type only the second one. The message with the type “lazy.pink.rabbit” will reach the second stage only once, although it corresponds to two connections. A message with the type “quick.brown.fox” will not reach any queue.

What will happen if you send a message with a key that does not correspond to any rule: with 1 or 4 words (for example, “quick.orange.male.rabbit”). Such a message will not come anywhere and retire.

But the message with the key “lazy.orange.male.rabbit, although it consists of 4 words, corresponds to the last connection and will fall into second place.

Topic is a very powerful access point. It can work like other access points. If you put „#“ in the binding key, it will behave like a fanout. And if you do not use the characters "*" and "#", then you will behave like direct.

Total we get

We will use the topic access point for our logging system. Let us assume that the routing key of the log contains 2 words: 'facility.severity'. The code is almost the same as in the previous article .

Script emit_log_topic.php (producer):

<?php require_once __DIR__ . '/vendor/autoload.php'; use PhpAmqpLib\Connection\AMQPConnection; use PhpAmqpLib\Message\AMQPMessage; $connection = new AMQPConnection('localhost', 5672, 'guest', 'guest'); $channel = $connection->channel(); $channel->exchange_declare('topic_logs', 'topic', false, false, false); $routing_key = $argv[1]; if(empty($routing_key)) $routing_key = "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(); ?> 

Script receive_logs_topic.php (subscriber):

 <?php require_once __DIR__ . '/vendor/autoload.php'; use PhpAmqpLib\Connection\AMQPConnection; $connection = new AMQPConnection('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(); ?> 

To get everything:

 $ php receive_logs_topic.php "#" 

To get all the logs of the kern object:

 $ phpreceive_logs_topic.php "kern.*" 

If you want to listen only logs with type critical:

 $ php receive_logs_topic.php "*.critical" 

Create multiple links:

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

To receive messages with the key "kern.critical":

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

Note that the code does not create any rules for the routing key and the binding key. You can try to enter into the key more than two words.

The complete script code is emit_log_topic.php and receive_logs_topic.php .