3.6.6 RabbitMQ教程五 – Topics

Posted 2021-03-15 infinitecodes

What This Tutorial Focuses On

In the previous tutorial we improved our logging system. Instead of using a fanout exchange only capable of dummy broadcasting, we used a direct one, and gained a possibility of selectively receiving the logs.

在之前的教程中我们改进了我们的日志系统。我们没有使用只能无脑广播的fanout交换，而是使用了direct交换，并获得了有选择的接收日志的可能性。

Although using the direct exchange improved our system, it still has limitations - it can‘t do routing based on multiple criteria.

虽然使用direct交换改进了我们的系统，但它仍然有局限性 - 它不能基于多个标准进行路由。

In our logging system we might want to subscribe to not only logs based on severity, but also based on the source which emitted the log. You might know this concept from the syslog unix tool, which routes logs based on both severity (info/warn/crit...) and facility (auth/cron/kern...).

在我们的日志系统中，我们可能希望不仅仅只订阅基于严重等级的日志，也订阅发送日志的源。

That would give us a lot of flexibility - we may want to listen to just critical errors coming from ‘cron‘ but also all logs from ‘kern‘.

这样会给我们很多灵活性 - 我们可能只想监听来自“cron”的关键错误，但也要监听来自“kern”的所有日志

To implement that in our logging system we need to learn about a more complex topic exchange.

要在我们的日志系统中实现那个我们需要了解更复杂的topic交换

Topic exchange

Messages sent to a topic exchange can‘t have an arbitrary routing_key - it must be a list of words, delimited by dots. The words can be anything, but usually they specify some features connected to the message. A few valid routing key examples: "stock.usd.nyse", "nyse.vmw", "quick.orange.rabbit". There can be as many words in the routing key as you like, up to the limit of 255 bytes.

发送至一个topic交换的消息不能有任意的routing_key - 它必须是一个单词列表，以点分隔。这些词可以是任何东西，但通常它们指定了与消息相关的一些特性。一些无效的routing key示例：‘stock.usd.nyse’, ‘nyse.vmw’, ‘quick.orange.rabbit’. routing key中可以有任意多个单词，最多255字节。

The binding key must also be in the same form. The logic behind the topic exchange is similar to a direct one - a message sent with a particular routing key will be delivered to all the queues that are bound with a matching binding key. However there are two important special cases for binding keys:

• * (star) can substitute for exactly one word.
• # (hash) can substitute for zero or more words.

binding key必须也是同样的形式。topic交换背后的逻辑与direct交换类似 - 一条带有特定的routing key的消息会被传递给所有绑定了相匹配的binding key的队列。但binding keys有两种重要的特殊情况

• *（星）只能替换一个单词
• #（井）可以替换零个或多个单词

It‘s easiest to explain this in an example:

在示例中解释这个最简单：

                                           

In this example, we‘re going to send messages which all describe animals. The messages will be sent with a routing key that consists of three words (two dots). The first word in the routing key will describe a celerity, second a colour and third a species: "<celerity>.<colour>.<species>".

在这个示例中，我们将发送所有描述动物的消息。消息会和一个routing key发送，routing key由三个单词（两个点）组成。routing key中的第一个单词会描述一个速度，第二个单词描述一种颜色然后第三个单词描述一个物种

We created three bindings: Q1 is bound with binding key "*.orange.*" and Q2 with "*.*.rabbit" and "lazy.#".

我们创建了三个bindings：Q1与binding key‘*.orange’绑定，并且Q2与‘*.*.rabbit’和‘lazy.#’绑定

These bindings can be summarised as:

• Q1 is interested in all the orange animals.
• Q2 wants to hear everything about rabbits, and everything about lazy animals.

这些bindings可以被总结为：

• Q1对所有橙色动物感兴趣
• Q2想要收听任何与rabbits有关的东西，还有所有关于懒惰的动物的东西。

A message with a routing key set to "quick.orange.rabbit" will be delivered to both queues. Message "lazy.orange.elephant" also will go to both of them. On the other hand "quick.orange.fox" will only go to the first queue, and "lazy.brown.fox" only to the second. "lazy.pink.rabbit" will be delivered to the second queue only once, even though it matches two bindings. "quick.brown.fox" doesn‘t match any binding so it will be discarded.

routing key设置为“quick.orange.rabbit”的消息将同时传递到两个队列。消息"lazy.orange.elephant"也会发送至两个队列。另一方面"quick.orange.fox"只会走第一个队列，然后"lazy.brown.fox"只会走第二个。"lazy.pink.rabbit"只会被传递至第二个队列一次，尽管它和两个bindings都匹配。"quick.brown.fox"不匹配任何binding所以它会被丢弃。

What happens if we break our contract and send a message with one or four words, like "orange" or "quick.orange.male.rabbit"? Well, these messages won‘t match any bindings and will be lost.

如果我们打破约定并发送一条有一个或四个单词的消息会发生什么呢？比如"orange"或者"quick.orange.male.rabbit"? 嗯，这些消息与任何bindings都不匹配，并且都会丢失

On the other hand "lazy.orange.male.rabbit", even though it has four words, will match the last binding and will be delivered to the second queue.

另一边"lazy.orange.male.rabbit", 尽管它有四个单词，但它和最后一个binding匹配并且会被传递至第二个队列

Topic exchange

Topic exchange is powerful and can behave like other exchanges.

Topic交换很强大并能像其他交换那样工作

When a queue is bound with "#" (hash) binding key - it will receive all the messages, regardless of the routing key - like in fanout exchange.

当一个队列绑定了一个为‘#’的binding key时 - 它会接收所有消息，不管routing key是什么 - 比如在fanout交换中。

When special characters "*" (star) and "#" (hash) aren‘t used in bindings, the topic exchange will behave just like a direct one.

当bindings不使用特殊字符‘*’和‘#’时，topic交换会像direct交换那样工作

Putting it all together

We‘re going to use a topic exchange in our logging system. We‘ll start off with a working assumption that the routing keys of logs will have two words: "<facility>.<severity>".

我们将会在我们的日志系统中使用一个topic交换。我们将从一个可行的假设开始，这个假设是日志的routing keys有两个单词：‘<facility>.<severity>’

The code is almost the same as in the previous tutorial.

示例的代码与之前教程的很相似

emit_log_topic.py

import pika
import sys

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host=‘localhost‘))
channel = connection.channel()

channel.exchange_declare(exchange=‘topic_logs‘, exchange_type=‘topic‘)

routing_key = sys.argv[1] if len(sys.argv) > 2 else ‘anonymous.info‘
message = ‘ ‘.join(sys.argv[2:]) or ‘Hello World!‘
channel.basic_publish(
    exchange=‘topic_logs‘, routing_key=routing_key, body=message)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()

receive_logs_topic.py

import pika
import sys

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host=‘localhost‘))
channel = connection.channel()

channel.exchange_declare(exchange=‘topic_logs‘, exchange_type=‘topic‘)

result = channel.queue_declare(‘‘, exclusive=True)
queue_name = result.method.queue

binding_keys = sys.argv[1:]
if not binding_keys:
    sys.stderr.write("Usage: %s [binding_key]...
" % sys.argv[0])
    sys.exit(1)

for binding_key in binding_keys:
    channel.queue_bind(
        exchange=‘topic_logs‘, queue=queue_name, routing_key=binding_key)

print(‘ [*] Waiting for logs. To exit press CTRL+C‘)


def callback(ch, method, properties, body):
    print(" [x] %r:%r" % (method.routing_key, body))


channel.basic_consume(
    queue=queue_name, on_message_callback=callback, auto_ack=True)

channel.start_consuming()

To receive all the logs run:

要接收所有日志运行：

python receive_logs_topic.py "#"

To receive all logs from the facility "kern":

要从设备‘kern’接收所有消息

python receive_logs_topic.py "kern.*"

Or if you want to hear only about "critical" logs:

或者也许你只想接收‘critical’日志：

python receive_logs_topic.py "*.critical"

You can create multiple bindings:

你可以创建多个bindings

python receive_logs_topic.py "kern.*" "*.critical"

And to emit a log with a routing key "kern.critical" type:

并发送一条带有名为‘kern.critical’的routing key的日志

python emit_log_topic.py "kern.critical" "A critical kernel error"

Have fun playing with these programs. Note that the code doesn‘t make any assumption about the routing or binding keys, you may want to play with more than two routing key parameters.

希望这些程序让你玩的愉快。注意，关于routing或binding keys，代码没有做任何假设。你也许会想试试多余两个关键参数的玩法