RabbitMQ消息队列（四）：分发到多Consumer（Publish/Subscribe）

上篇文章中，咱们把每一个Message都是deliver到某个Consumer。在这篇文章中，咱们将会将同一个Message deliver到多个Consumer中。这个模式也被成为 “publish / subscribe”。

这篇文章中，咱们将建立一个日志系统，它包含两个部分：第一个部分是发出log（Producer），第二个部分接收到并打印（Consumer）。 咱们将构建两个Consumer，第一个将log写到物理磁盘上；第二个将log输出的屏幕。

1. Exchanges

关于exchange的概念在《RabbitMQ消息队列（一）: Detailed Introduction 详细介绍》中有详细介绍。如今作一下简单的回顾。

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：

channel.exchange_declare(exchange='logs',
                         type='fanout')复制代码

Listing exchanges

经过rabbitmqctl能够列出当前全部的Exchange：

$ sudo rabbitmqctl list_exchanges
Listing exchanges ...
logs      fanout
amq.direct      direct
amq.topic       topic
amq.fanout      fanout
amq.headers     headers
...done.复制代码

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

如今咱们能够经过exchange，而不是routing_key来publish Message了：

channel.basic_publish(exchange='logs',
                      routing_key='',
                      body=message)复制代码

2. Temporary queues

截至如今，咱们用的queue都是有名字的：第一个是hello，第二个是task_queue。使用有名字的queue，使得在Producer和Consumer以前共享queue成为可能。

可是对于咱们将要构建的日志系统，并不须要有名字的queue。咱们但愿获得全部的log，而不是它们中间的一部分。并且咱们只对当前的log感兴趣。为了实现这个目标，咱们须要两件事情：
1） 每当Consumer链接时，咱们须要一个新的，空的queue。由于咱们不对老的log感兴趣。幸运的是，若是在声明queue时不指定名字，那么RabbitMQ会随机为咱们选择这个名字。方法：

result = channel.queue_declare()复制代码

经过result.method.queue 能够取得queue的名字。基本上都是这个样子：amq.gen-JzTY20BRgKO-HjmUJj0wLg。
2）当Consumer关闭链接时，这个queue要被deleted。能够加个exclusive的参数。方法：

result = channel.queue_declare(exclusive=True)复制代码

复制代码

3. Bindings绑定

如今咱们已经建立了fanout类型的exchange和没有名字的queue（其实是RabbitMQ帮咱们取了名字）。那exchange怎么样知道它的Message发送到哪一个queue呢？答案就是经过bindings：绑定。

方法：

channel.queue_bind(exchange='logs',
                   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:

#!/usr/bin/env python
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!"
channel.basic_publish(exchange='logs',
                      routing_key='',
                      body=message)
print " [x] Sent %r" % (message,)
connection.close()复制代码

还有一点要注意的是咱们声明了exchange。publish到一个不存在的exchange是被禁止的。若是没有queue bindings exchange的话，log是被丢弃的。
Consumer：receive_logs.py:

#!/usr/bin/env python
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()复制代码

咱们开始不是说须要两个Consumer吗？一个负责记录到文件；一个负责打印到屏幕？
其实用重定向就能够了，固然你想修改callback本身写文件也行。咱们使用重定向的方法：
We’re done. If you want to save logs to a file, just open a console and type:

$ python receive_logs.py > logs_from_rabbit.log复制代码

Consumer2：打印到屏幕：

$ python receive_logs.py复制代码

接下来，Producer：

$ python emit_log.py复制代码

使用命令rabbitmqctl list_bindings你能够看咱们建立的queue。
一个output：

$ sudo rabbitmqctl list_bindings
Listing bindings ...
logs    exchange        amq.gen-JzTY20BRgKO-HjmUJj0wLg  queue           []
logs    exchange        amq.gen-vso0PVvyiRIL2WoV3i48Yg  queue           []
...done.复制代码

这个结果仍是很好理解的。

参考资料：

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

2. http://blog.csdn.net/anzhsoft/article/details/19617305