跳至主要内容

RabbitMQ 教程 - 主题

主题​

(使用 Pika Python 客户端)​

信息

先决条件​

本教程假定 RabbitMQ 已安装并正在运行，位于localhost上的标准端口（5672）。如果您使用不同的主机、端口或凭据，则连接设置需要调整。

在哪里获取帮助​

如果您在完成本教程时遇到问题，可以通过GitHub 讨论或RabbitMQ 社区 Discord与我们联系。

先决条件​

与其他 Python 教程一样，我们将使用Pika RabbitMQ 客户端版本 1.0.0。

本教程重点介绍的内容​

在上一个教程中，我们改进了日志记录系统。我们不再使用只能进行虚拟广播的fanout交换机，而是使用direct交换机，并获得了选择性接收日志的可能性。

虽然使用direct交换机改进了我们的系统，但它仍然存在局限性——它无法基于多个条件进行路由。

在我们的日志记录系统中，我们可能希望不仅订阅基于严重性的日志，还希望订阅基于发出日志的源的日志。您可能从syslog unix 工具中了解到此概念，该工具基于严重性（info/warn/crit…）和设施（auth/cron/kern…）对日志进行路由。

这将给我们带来很大的灵活性——我们可能希望监听来自“cron”的严重错误，但也希望监听来自“kern”的所有日志。

为了在我们的日志记录系统中实现这一点，我们需要了解更复杂的topic交换机。

主题交换机​

发送到topic交换机的消息不能具有任意的routing_key——它必须是点分隔的单词列表。这些单词可以是任何内容，但通常它们指定与消息相关的某些功能。一些有效的路由键示例：stock.usd.nyse、nyse.vmw、quick.orange.rabbit。路由键中可以包含任意数量的单词，最多 255 字节。

绑定键也必须采用相同的形式。topic交换机背后的逻辑类似于direct交换机——使用特定路由键发送的消息将被传递到所有使用匹配的绑定键进行绑定的队列。但是，绑定键有两个重要的特殊情况

  • *（星号）可以替代正好一个单词。
  • #（井号）可以替代零个或多个单词。

用一个例子解释起来最容易

在此示例中，我们将发送描述动物的所有消息。消息将使用包含三个单词（两个点）的路由键发送。路由键中的第一个单词将描述速度，第二个单词描述颜色，第三个单词描述物种：<celerity>.<colour>.<species>。

我们创建了三个绑定：Q1 使用绑定键*.orange.*绑定，Q2 使用*.*.rabbit和lazy.#绑定。

这些绑定可以概括为

  • Q1 对所有橙色动物感兴趣。
  • Q2 想要听到关于兔子的一切，以及关于懒惰动物的一切。

具有路由键quick.orange.rabbit的消息将被传递到这两个队列。消息lazy.orange.elephant也将被传递到这两个队列。另一方面，quick.orange.fox只会发送到第一个队列，lazy.brown.fox只会发送到第二个队列。lazy.pink.rabbit将只发送到第二个队列一次，即使它匹配两个绑定。quick.brown.fox不匹配任何绑定，因此将被丢弃。

如果我们违反协议，发送一个或四个单词的消息，例如orange或quick.orange.new.rabbit会发生什么？好吧，这些消息将不匹配任何绑定，并将丢失。

另一方面，lazy.orange.new.rabbit，即使它有四个单词，也会匹配最后一个绑定，并将被传递到第二个队列。

主题交换机​

主题交换机功能强大，可以像其他交换机一样工作。

当一个队列使用#（井号）绑定键进行绑定时，它将接收所有消息，无论路由键是什么——就像在fanout交换机中一样。

当在绑定中不使用特殊字符*（星号）和#（井号）时，主题交换机将像direct交换机一样工作。

将所有内容整合在一起​

我们将在日志记录系统中使用topic交换机。我们将从一个工作假设开始，即日志的路由键将有两个单词：<facility>.<severity>。

代码几乎与上一个教程中的代码相同。

emit_log_topic.py (源代码)

#!/usr/bin/env python
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(f" [x] Sent {routing_key}:{message}")
connection.close()

receive_logs_topic.py (源代码)

#!/usr/bin/env python
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]...\n" % 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(f" [x] {method.routing_key}:{body}")


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

channel.start_consuming()

要接收所有日志，请运行

python receive_logs_topic.py "#"

要接收来自设施kern的所有日志，请运行

python receive_logs_topic.py "kern.*"

或者，如果您只想听取critical日志，请运行

python receive_logs_topic.py "*.critical"

您可以创建多个绑定

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

要发出具有路由键kern.critical类型的日志，请键入

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

玩转这些程序吧。请注意，代码没有对路由或绑定键进行任何假设，您可能希望使用超过两个路由键参数进行玩转。

继续学习教程 6，了解RPC。

© . All rights reserved.