1. 簡介

上節我們提到直連交換機，並基於我們的日誌系統使用直連交換機，而本節我們使用rabbitmq內建的另外一種交換機——主題交換機。
所謂主題交換機是什麼呢？通俗點來講就是直連交換機的升級版，我們可以認為直連交換機就是她其中的一種形式。
直連交換機需要通過繫結（bindings）和佇列建立起關係，而其主要通過的就是routing_key，並且routing_key是要與生產者提供的routing_key完全匹配才能得到訊息。主題交換機對其進行了升級，升級的方式就是通過兩個符號"#"和"*"以及對routing_key的長度限制和格式限制，完成了模糊匹配，從而達到消費者可以更靈活的匹配要處理的訊息，下面再具體針對例項來進行主題交換機的學習。

2. 主題交換機

傳送到主題交換機（topic exchange）的訊息攜帶的路由鍵（routing_key）必須是一個由.分隔開的詞語列表。這些單詞隨便是什麼都可以，但是最好是跟攜帶它們的訊息有關係的詞彙。以下是幾個推薦的例子："stock.usd.nyse", "nyse.vmw", "quick.orange.rabbit"。詞語的個數可以隨意，但是不要超過255位元組。
繫結鍵也必須擁有同樣的格式。主題交換機背後的邏輯跟直連交換機很相似 —— 一個攜帶著特定路由鍵的訊息會被主題交換機投遞給繫結鍵與之想匹配的佇列。但是它的繫結鍵和路由鍵有兩個特殊應用方式：

"*" (星號) 用來表示一個單詞。
"#"

(井號) 用來表示任意數量（零個或多個）單詞。

大家可以根據下圖，猜測哪些路由鍵會被髮送到相應的佇列，就是類似於正則的匹配，但是比正則簡單太多了，有興趣的可以到官網檢視完整內容，我就不再闡述了。

主題交換機

當一個佇列的繫結鍵為 "#"（井號） 的時候，這個佇列將會無視訊息的路由鍵，接收所有的訊息。
當 * (星號) 和 # (井號) 這兩個特殊字元都未在繫結鍵中出現的時候，此時主題交換機就擁有的直連交換機的行為。
所以主題交換機也就實現了扇形交換機的功能，和直連交換機的功能

總結

我們基於整個流程進行一下總結

生產者

建立連線
建立交換機（主題交換機）
傳送訊息到指定交換機，並將訊息中指定routing_key（注意key的規則）

消費者

建立連線
建立交換機（主題交換機）
建立匿名佇列（每個消費者都要建立的獨立的佇列，用於接收路由後訊息）
建立繫結（通過路由鍵繫結交換機佇列關係，確定佇列處理交換機中的哪些訊息，注意路由鍵的正則規則）
在指定交換機處理訊息

程式碼整合

emit_logs_topic.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Date    : 2016-02-28 21:28:17
# @Author  : mx ([email protected])
# @Link    : http://www.shujutiyu.com/
# @Version : $Id$

import os
import pika
import sys

conn = None

routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'

try:
    # 獲取連線
    conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    # 獲取通道
    channel = conn.channel()

    # 建立直連交換機
    channel.exchange_declare(exchange='topic_logs',
                             type='topic')

    # 在RabbitMQ中傳送訊息，指定交換機（exchange），指定routing
    ret = channel.basic_publish(exchange='topic_logs',
                                routing_key=routing_key,
                                body=message,)
    print " [x] Sent '{0}'".format(message)
    print ret
except Exception, e:
    raise e
finally:
    if conn:
        conn.close()

recv_logs_topic.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Date    : 2016-02-29 16:30:21
# @Author  : mx ([email protected])
# @Link    : http://www.shujutiyu.com/
# @Version : $Id$

import pika
import sys

conn = None

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

def callback(ch, method, properties, body):
    """
        @ch: channel 通道，是由外部呼叫時上送
        out body
        讀取佇列內容做不同的操作
    """
    print " [x] method.routing_key {0}".format(method.routing_key)
    print " [x] Done %r" % (body, )

try:
    # get connection
    conn = pika.BlockingConnection(pika.ConnectionParameters(
        'localhost')
    )

    # get channel
    channel = conn.channel()

    # 宣告直連交換機
    channel.exchange_declare(exchange='topic_logs', 
                             type='topic')

    # 宣告臨時佇列 , param exclusive 互斥
    tmp_queue = channel.queue_declare(exclusive=True)
    queue_name = tmp_queue.method.queue

    # 根據輸入routing_key繫結交換機與佇列
    for binding_key in binding_keys:
        channel.queue_bind( exchange='topic_logs', 
                            queue=queue_name,
                            routing_key=binding_key )

    # 在佇列中讀取資訊
    channel.basic_consume(callback, queue=queue_name, no_ack=True)

    print ' [*] Waiting for messages. To exit press CTRL+C'

    channel.start_consuming()
except Exception, e:
    raise e
finally:
    if conn:
        conn.close()

疑問

• 繫結鍵為 * 的佇列會取到一個路由鍵為空的訊息嗎？
• 繫結鍵為 #.* 的佇列會獲取到一個名為..的路由鍵的訊息嗎？它會取到一個路由鍵為單個單詞的訊息嗎？
• a.*.# 和 a.#的區別在哪兒？
• 之前提到了佇列、訊息、交換機可以持久化，有興趣的不妨試一下

3. 參考資料