源碼：https://github.com/ltoddy/rabbitmq-tutorial

Topics

(using the Pika Python client)

本章節教程重點介紹的內容

在之前的教程中，我們改進了日志記錄系統。我們沒有使用只有虛擬廣播的fanout交換，而是使用了direct交換，并讓選擇性接收日志成為了可能。

盡管使用direct交換改進了我們的系統，但它仍然有局限性 - 它不能根據多個標準進行路由。

在我們的日志系統中，我們可能不僅需要根據嚴重性來訂閱日志，還要根據發布日志的來源進行訂閱。您可能從syslog unix工具知道這個概念，
該工具根據嚴重性(info / warning / crit...)和工具(auth / cron / kern ...)來路由日志。

這會給我們很大的靈活性 - 因為我們可能想聽取來自'cron'的error日志，而且還聽取來自'kern'的所有日志。

為了在我們的日志系統中實現這一點，我們需要了解更復雜的topic交換。

Topic 交換

發送到topic交換的消息必須有規范的routing_key - 它必須是由點分隔的單詞列表。單詞可以是任何東西，但通常它們指定了與該消息相關的一些功能。
一些有效的routing_key例子: "stock.usd.nyse"，"nyse.vmw"，"quick.orange.rabbit"。只要您愿意，路由鍵中可以有任意的單詞，但最多255個字節。

綁定鍵也必須是相同的形式。topic交換背后的邏輯與direct topic交換背后的邏輯類似 - 使用特定路由鍵發送的消息將被傳遞到與匹配綁定鍵綁定的所有隊列。
但是綁定鍵有兩個重要的特殊情況：

• * (star) 可以代替一個字。
• # (hash) 可以替代零個或多個單詞。

在這個例子中解釋這個很簡單：

在這個例子中，我們將發送所有描述動物的消息。消息將使用由三個字（兩個點）組成的路由鍵發送。
路由關鍵字中的第一個單詞將描述速度，第二個顏色和第三個物種：" <celerity> <color> <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.male.rabbit"，
會發生什么情況？那么，這些消息將不匹配任何綁定，并會丟失。

另一方面，"lazy.orange.male.rabbit"即使有四個單詞，也會匹配最后一個綁定，并將傳遞到第二個隊列。

direct change話題交換功能強大，可以像其他交流一樣行事。當使用" \# "（散列）綁定鍵綁定隊列時，它將接收所有消息，而不管路由密鑰如何 - 就像在*fanout*交換中一樣。當在綁定中沒有使用特殊字符"\*"(星號)和"\#"(散列)時，主題交換將像*direct*交換一樣。

把它放在一起

我們將在我們的日志系統中使用topic交換。我們首先假定日志的路由鍵有兩個單詞：" <facility>.<severity> "。

代碼幾乎與前一個教程中的代碼相同 。

emit_log_topic.py的代碼：

#!/usr/bin/env python import sys import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('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的代碼:

#!/usr/bin/env python import sys import pika# connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) connection = pika.BlockingConnection(pika.ConnectionParameters('172.17.0.2')) channel = connection.channel()channel.exchange_declare(exchange='topic_logs',exchange_type='topic')result = channel.queue_declare(exclusive=True) queue_name = result.method.queuebinding_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(" [x] %r:%r" % (method.routing_key, body))channel.basic_consume(callback,queue=queue_name,no_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" 創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

以上是生活随笔為你收集整理的rabbitmq中文教程python版 - Topics的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。