RabbitMQ（python實現）學習之二：Producer發送消息至多個消息隊列queue（廣播消息）

1.1本部分內容簡介

這部分我們將要發送一個消息到多個Consumer，這部分稱之為“publish/subscribe”

我們實現的方式就是發送端，發送一個消息，與此同時，多個接收端將同時接收到消息並打印在屏幕上面。

1.2exchange簡介

在前面的博文中，我們的講解是：發送端發送消息至消息隊列，接收端從消息隊列獲取消息。現在我們來介紹一下rabbitmq的完整消息傳送模型。

>Producer：用來發送消息的應用程序

>queue：用來存儲消息的緩存

>Consumer：用來接收消息的應用程序

消息傳送模型的核心是，Producer從不會直接將消息傳送給queue，而是，將消息傳送給exchange，exchange是個很簡單的東西，在一側，他接收來自Producer的消息，另一側將消息傳送給queue。exchange將消息傳送給你個queue，還是傳送給多個queue，這主要是由exchange的type決定。模型圖如下：

　　　　　　　　

 

exchange有很多type可用，如下：direct、topic、headers、fanout。本博客針對fanout講解，后續博文對其他類型有所講解，讓我們創建一個exchange，type為fanout，名字為logs，代碼如下：

channel.exchange(exchange='logs',type='fanout')

對於type為fanout的exchange，理解起來非常簡單，它將接收到的消息，廣播給他所知道的所有的queue，即所有和他建立連接的queue。前面的博文降到了命令行查看list_exchanges的命令如下：

 $ ：sudo rabbitmqctl list_exchanges
 Listing exchanges ...
 logs      fanout
 amq.direct      direct
 amq.topic       topic
 amq.fanout      fanout
 amq.headers     headers
 ...done.

對於上圖中，你會看到很多amq.*的exchange，這些是系統默認建立的，在你不建立exchange時，系統默認建立上面幾個。

對於消息的發布函數basic_publish（）也隨之變為：

channel.basic_publish(exchange='logs',routing_key='',body=message)

1.3臨時queue

 正如你前面學到的，對於一個queue，會有自己的名字（hello什么的），

首先：

result = channel.queue_declare()

然后通過result.method.queue，系統會隨機給queue命名。

如果我們想Producer與Consumer斷開連接時，隊列queue刪除，那么需要改成下面的代碼：

result = channel.queue_declare(exclusive=True)

1.4Bingings(將queue與exchange綁定)

模型圖如下：

　　　　　　　　

我們已經創建了一個type為fanout的exchange，現在，我們要告訴exchange，將消息發送給我們自己定義的queue，在exchange與queue之間建立連接的是binding，代碼如下：

channel.queue_bind(exchange='logs',queue=result.method.queue)

在命令行查看binding的列表，命令如下：

$: sudo rabbitmqctl list_bindings

 1.5最終代碼

最終的模型如下：

　　　　　　　　   

send.py代碼如下：

import pika
import sys

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

channel.exchange_declare(exchange='logs',
                         type='fanout')

message = ' '.join(sys.argv[1:]) or "info: Hello World!"  #如果鍵盤有輸入，message為鍵盤輸入，如果鍵盤沒有輸入，消息message="info: Hello World!"；
 channel.basic_publish(exchange='logs', routing_key='', body=message) print " [x] Sent %r" % (message,) connection.close()

 

receive.py代碼

import pika

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

channel.exchange_declare(exchange='logs',
                         type='fanout')

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

channel.queue_bind(exchange='logs',
                   queue=queue_name)

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

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

channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)

channel.start_consuming()

1.6代碼測試

開啟一個命令行窗口，運行send.py：

$: python send.py   #(此時你傳送的內容為info: Hello World!)或者  

$: python send.py message #message為你想發送的內容

開啟兩個命令行窗口，分別運行receive.py，兩個窗口你會看到有相同的消息輸出：

$: python receive.py