<=== RabbitMQ訊息佇列（三）：任務分發機制

      上篇文章中，我們把每個Message都是deliver到某個Consumer。在這篇文章中，我們將會將同一個Message deliver到多個Consumer中。這個模式也被成為 "publish / subscribe"。
    這篇文章中，我們將建立一個日誌系統，它包含兩個部分：第一個部分是發出log（Producer），第二個部分接收到並列印（Consumer）。 我們將構建兩個Consumer，第一個將log寫到物理磁碟上；第二個將log輸出的螢幕。

1. Exchanges

    RabbitMQ 的Messaging Model就是Producer並不會直接傳送Message到queue。實際上，Producer並不知道它傳送的Message是否已經到達queue。

   Producer傳送的Message實際上是發到了Exchange中。它的功能也很簡單：從Producer接收Message，然後投遞到queue中。Exchange需要知道如何處理Message，是把它放到那個queue中，還是放到多個queue中？這個rule是通過Exchange 的型別定義的。

   我們知道有三種類型的Exchange：direct, topic 和fanout。fanout就是廣播模式，會將所有的Message都放到它所知道的queue中。建立一個名字為logs，型別為fanout的Exchange：

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

Listing exchanges

通過rabbitmqctl可以列出當前所有的Exchange：

1. $ sudo rabbitmqctl list_exchanges  
2. Listing exchanges ...  
3. logs      fanout  
4. amq.direct      direct  
5. amq.topic       topic  
6. amq.fanout      fanout  
7. amq.headers     headers  
8. ...done.  

注意 amq.* exchanges 和the default (unnamed)exchange是RabbitMQ預設建立的。

現在我們可以通過exchange，而不是routing_key來publish Message了：

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

2. Temporary queues

    截至現在，我們用的queue都是有名字的：第一個是hello，第二個是task_queue。使用有名字的queue，使得在Producer和Consumer之前共享queue成為可能。

    但是對於我們將要構建的日誌系統，並不需要有名字的queue。我們希望得到所有的log，而不是它們中間的一部分。而且我們只對當前的log感興趣。為了實現這個目標，我們需要兩件事情：
    1） 每當Consumer連線時，我們需要一個新的，空的queue。因為我們不對老的log感興趣。幸運的是，如果在宣告queue時不指定名字，那麼RabbitMQ會隨機為我們選擇這個名字。方法：

1. result = channel.queue_declare()  

    通過result.method.queue 可以取得queue的名字。基本上都是這個樣子：amq.gen-JzTY20BRgKO-HjmUJj0wLg。
    2）當Consumer關閉連線時，這個queue要被deleted。可以加個exclusive的引數。方法：

1. result = channel.queue_declare(exclusive=True)  

3. Bindings繫結

現在我們已經建立了fanout型別的exchange和沒有名字的queue（實際上是RabbitMQ幫我們取了名字）。那exchange怎麼樣知道它的Message傳送到哪個queue呢？答案就是通過bindings：繫結。

方法：

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

現在logs的exchange就將它的Message附加到我們建立的queue了。 Listing bindings

使用命令rabbitmqctl list_bindings。

4. 最終版本

    我們最終實現的資料流圖如下：

Producer，在這裡就是產生log的program，基本上和前幾個都差不多。最主要的區別就是publish通過了exchange而不是routing_key。

emit_log.py script:

1. #!/usr/bin/env python
2. import pika  
3. import sys  
4. connection = pika.BlockingConnection(pika.ConnectionParameters(  
5.         host='localhost'))  
6. channel = connection.channel()  
7. channel.exchange_declare(exchange='logs',  
8.                          type='fanout')  
9. message = ' '.join(sys.argv[1:]) or"info: Hello World!"
10. channel.basic_publish(exchange='logs',  
11.                       routing_key='',  
12.                       body=message)  
13. print" [x] Sent %r" % (message,)  
14. connection.close()  

還有一點要注意的是我們聲明瞭exchange。publish到一個不存在的exchange是被禁止的。如果沒有queue bindings exchange的話，log是被丟棄的。
Consumer：receive_logs.py:

1. #!/usr/bin/env python
2. import pika  
3. connection = pika.BlockingConnection(pika.ConnectionParameters(  
4.         host='localhost'))  
5. channel = connection.channel()  
6. channel.exchange_declare(exchange='logs',  
7.                          type='fanout')  
8. result = channel.queue_declare(exclusive=True)  
9. queue_name = result.method.queue  
10. channel.queue_bind(exchange='logs',  
11.                    queue=queue_name)  
12. print' [*] Waiting for logs. To exit press CTRL+C'
13. def callback(ch, method, properties, body):  
14.     print" [x] %r" % (body,)  
15. channel.basic_consume(callback,  
16.                       queue=queue_name,  
17.                       no_ack=True)  
18. channel.start_consuming()  

我們開始不是說需要兩個Consumer嗎？一個負責記錄到檔案；一個負責列印到螢幕？
其實用重定向就可以了，當然你想修改callback自己寫檔案也行。我們使用重定向的方法：
We're done. If you want to save logs to a file, just open a console and type:

1. $ python receive_logs.py > logs_from_rabbit.log  

Consumer2：列印到螢幕：

1. $ python receive_logs.py  

接下來，Producer：

1. $ python emit_log.py  

使用命令rabbitmqctl list_bindings你可以看我們建立的queue。一個output：

1. $ sudo rabbitmqctl list_bindings  
2. Listing bindings ...  
3. logs    exchange        amq.gen-JzTY20BRgKO-HjmUJj0wLg  queue           []  
4. logs    exchange        amq.gen-vso0PVvyiRIL2WoV3i48Yg  queue           []  
5. ...done.  

這個結果還是很好理解的。

尊重原創，轉載請註明出處 anzhsoft： http://blog.csdn.net/anzhsoft/article/details/19617305

參考資料：

1. http://www.rabbitmq.com/tutorials/tutorial-three-python.html

 <=== RabbitMQ訊息佇列（三）：任務分發機制

      上篇文章中，我們把每個Message都是deliver到某個Consumer。在這篇文章中，我們將會將同一個Message deliver到多個Consumer中。這個模式也被成為 "publish / subscribe"。
    這篇文章中，我們將建立一個日誌系統，它包含兩個部分：第一個部分是發出log（Producer），第二個部分接收到並列印（Consumer）。 我們將構建兩個Consumer，第一個將log寫到物理磁碟上；第二個將log輸出的螢幕。

1. Exchanges

    RabbitMQ 的Messaging Model就是Producer並不會直接傳送Message到queue。實際上，Producer並不知道它傳送的Message是否已經到達queue。

   Producer傳送的Message實際上是發到了Exchange中。它的功能也很簡單：從Producer接收Message，然後投遞到queue中。Exchange需要知道如何處理Message，是把它放到那個queue中，還是放到多個queue中？這個rule是通過Exchange 的型別定義的。

   我們知道有三種類型的Exchange：direct, topic 和fanout。fanout就是廣播模式，會將所有的Message都放到它所知道的queue中。建立一個名字為logs，型別為fanout的Exchange：

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

Listing exchanges

通過rabbitmqctl可以列出當前所有的Exchange：

1. $ sudo rabbitmqctl list_exchanges  
2. Listing exchanges ...  
3. logs      fanout  
4. amq.direct      direct  
5. amq.topic       topic  
6. amq.fanout      fanout  
7. amq.headers     headers  
8. ...done.  

注意 amq.* exchanges 和the default (unnamed)exchange是RabbitMQ預設建立的。

現在我們可以通過exchange，而不是routing_key來publish Message了：

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

2. Temporary queues

    截至現在，我們用的queue都是有名字的：第一個是hello，第二個是task_queue。使用有名字的queue，使得在Producer和Consumer之前共享queue成為可能。

    但是對於我們將要構建的日誌系統，並不需要有名字的queue。我們希望得到所有的log，而不是它們中間的一部分。而且我們只對當前的log感興趣。為了實現這個目標，我們需要兩件事情：
    1） 每當Consumer連線時，我們需要一個新的，空的queue。因為我們不對老的log感興趣。幸運的是，如果在宣告queue時不指定名字，那麼RabbitMQ會隨機為我們選擇這個名字。方法：

1. result = channel.queue_declare()  

    通過result.method.queue 可以取得queue的名字。基本上都是這個樣子：amq.gen-JzTY20BRgKO-HjmUJj0wLg。
    2）當Consumer關閉連線時，這個queue要被deleted。可以加個exclusive的引數。方法：

1. result = channel.queue_declare(exclusive=True)  

3. Bindings繫結

現在我們已經建立了fanout型別的exchange和沒有名字的queue（實際上是RabbitMQ幫我們取了名字）。那exchange怎麼樣知道它的Message傳送到哪個queue呢？答案就是通過bindings：繫結。

方法：

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

現在logs的exchange就將它的Message附加到我們建立的queue了。 Listing bindings

使用命令rabbitmqctl list_bindings。

4. 最終版本

    我們最終實現的資料流圖如下：