4.rabbitmq消息队列(四):分发到多consumer(publish、subscribe)

anjie2017 贡献于2017-01-06

作者 Administrator  创建于2014-10-29 12:08:00   修改者Administrator  修改于2017-01-04 08:18:48字数3822

文档摘要:在这篇文章中,我们将会将同一个Messagedeliver到多个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: [python]view plaincopy 1. channel.exchange_declare(exchange='logs',   2.                          type='fanout')   Listing exchanges 通过rabbitmqctl可以列出当前所有的Exchange: [python]view plaincopy 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了: [python]view plaincopy 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会随机为我们选择这个名字。方法: [python]view plaincopy 1. result = channel.queue_declare()       通过result.method.queue 可以取得queue的名字。基本上都是这个样子:amq.gen-JzTY20BRgKO-HjmUJj0wLg。     2)当Consumer关闭连接时,这个queue要被deleted。可以加个exclusive的参数。方法: [python]view plaincopy 1. result = channel.queue_declare(exclusive=True)   3. Bindings绑定 现在我们已经创建了fanout类型的exchange和没有名字的queue(实际上是RabbitMQ帮我们取了名字)。那exchange怎么样知道它的Message发送到哪个queue呢?答案就是通过bindings:绑定。 方法: [python]view plaincopy 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: [python]view plaincopy 1. #!/usr/bin/env python   2. import pika   3. import sys   4.    5. connection = pika.BlockingConnection(pika.ConnectionParameters(   6.         host='localhost'))   7. channel = connection.channel()   8.    9. channel.exchange_declare(exchange='logs',   10.                          type='fanout')   11.    12. message = ' '.join(sys.argv[1:]) or "info: Hello World!"   13. channel.basic_publish(exchange='logs',   14.                       routing_key='',   15.                       body=message)   16. print " [x] Sent %r" % (message,)   17. connection.close()   还有一点要注意的是我们声明了exchange。publish到一个不存在的exchange是被禁止的。如果没有queue bindings exchange的话,log是被丢弃的。 Consumer:receive_logs.py: [python]view plaincopy 1. #!/usr/bin/env python   2. import pika   3.    4. connection = pika.BlockingConnection(pika.ConnectionParameters(   5.         host='localhost'))   6. channel = connection.channel()   7.    8. channel.exchange_declare(exchange='logs',   9.                          type='fanout')   10.    11. result = channel.queue_declare(exclusive=True)   12. queue_name = result.method.queue   13.    14. channel.queue_bind(exchange='logs',   15.                    queue=queue_name)   16.    17. print ' [*] Waiting for logs. To exit press CTRL+C'   18.    19. def callback(ch, method, properties, body):   20.     print " [x] %r" % (body,)   21.    22. channel.basic_consume(callback,   23.                       queue=queue_name,   24.                       no_ack=True)   25.    26. 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 1. $ python receive_logs.py > logs_from_rabbit.log   Consumer2:打印到屏幕: [python]view plaincopy 1. $ python receive_logs.py   接下来,Producer: [python]view plaincopy 1. $ python emit_log.py   使用命令rabbitmqctl list_bindings你可以看我们创建的queue。 一个output: [python]view plaincopy 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.   这个结果还是很好理解的。

下载文档到电脑,查找使用更方便

文档的实际排版效果,会与网站的显示效果略有不同!!

需要 10 金币 [ 分享文档获得金币 ] 0 人已下载

下载文档