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

 <=== RabbitMQ消息队列（三）：任务分发机制     

      上篇文章中，我们把每个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：

[python] view
plaincopy





channel.exchange_declare(exchange='logs',  

                         type='fanout')  

Listing exchanges

通过rabbitmqctl可以列出当前所有的Exchange：

[python] view
plaincopy





$ 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了：

[python] view
plaincopy





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会随机为我们选择这个名字。方法：

[python] view
plaincopy





result = channel.queue_declare()  

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

[python] view
plaincopy





result = channel.queue_declare(exclusive=True)  

3. Bindings绑定

现在我们已经创建了fanout类型的exchange和没有名字的queue（实际上是RabbitMQ帮我们取了名字）。那exchange怎么样知道它的Message发送到哪个queue呢？答案就是通过bindings：绑定。


方法：

[python] view
plaincopy





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:

[python] view
plaincopy





#!/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:

[python] view
plaincopy





#!/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] view
plaincopy





$ python receive_logs.py > logs_from_rabbit.log  

Consumer2：打印到屏幕：

[python] view
plaincopy





$ python receive_logs.py  

接下来，Producer：

[python] view
plaincopy





$ python emit_log.py  

使用命令rabbitmqctl list_bindings你可以看我们创建的queue。一个output：

[python] view
plaincopy





$ sudo rabbitmqctl list_bindings  

Listing bindings ...  

logs    exchange        amq.gen-JzTY20BRgKO-HjmUJj0wLg  queue           []  

logs    exchange        amq.gen-vso0PVvyiRIL2WoV3i48Yg  queue           []  

...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

    关于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：

[python] view
plaincopy





channel.exchange_declare(exchange='logs',  

                         type='fanout')  

Listing exchanges

通过rabbitmqctl可以列出当前所有的Exchange：

[python] view
plaincopy





$ 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了：

[python] view
plaincopy





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会随机为我们选择这个名字。方法：

[python] view
plaincopy





result = channel.queue_declare()  

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

[python] view
plaincopy





result = channel.queue_declare(exclusive=True)  

3. Bindings绑定

现在我们已经创建了fanout类型的exchange和没有名字的queue（实际上是RabbitMQ帮我们取了名字）。那exchange怎么样知道它的Message发送到哪个queue呢？答案就是通过bindings：绑定。


方法：

[python] view
plaincopy





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:

[python] view
plaincopy





#!/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:

[python] view
plaincopy





#!/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] view
plaincopy





$ python receive_logs.py > logs_from_rabbit.log  

Consumer2：打印到屏幕：

[python] view
plaincopy





$ python receive_logs.py  

接下来，Producer：

[python] view
plaincopy





$ python emit_log.py  

使用命令rabbitmqctl list_bindings你可以看我们创建的queue。

一个output：

[python] view
plaincopy





$ sudo rabbitmqctl list_bindings  

Listing bindings ...  

logs    exchange        amq.gen-JzTY20BRgKO-HjmUJj0wLg  queue           []  

logs    exchange        amq.gen-vso0PVvyiRIL2WoV3i48Yg  queue           []  

...done.  

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

尊重原创，转载请注明出处 anzhsoft： http://blog.csdn.net/anzhsoft/article/details/19617305


参考资料：

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