zoukankan      html  css  js  c++  java
  • 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中。exchange的功能也很简单:从Producer接收Message,然后投递到queue中。Exchange需要知道如何处理Message,是把它放到那个queue中,还是放到多个queue中?这个rule是通过Exchange 的类型定义的。

    流程:

      1)producer创建exchange,并且把message发送到exchange上

      2)consumer创建同一个exchange,创建一个临时的queue,并且通过bind把临时queue绑定到这个exchange上

      3)consumer订阅这个临时queue,也就是接收这个临时queue中的消息,这样producer就可以把消息发送给consumer

       我们知道有三种类型的Exchange:direct, topic 和fanout。fanout就是广播模式,会将所有的Message都放到它所知道的queue中。创建一个名字为logs,类型为fanout的Exchange:

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

    通过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)
    

      

      截至现在,我们用的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:

    Consumer1:打印到文件:

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

     

  • 相关阅读:
    Sharepoint 2010 使用asp.net web应用程序,调试sharepoint程序 解决办法
    sharepoint 管理中心 修改场管理员密码
    Sharepoint 给文档添加评论功能
    Sharepoint 修改评分列数据显示图标的颜色 (等级)
    sharepoint 启用评分功能
    sharepoint 获取术语集源的术语并绑定到下拉列表中
    逆元 板子
    CodeForces
    SPOJ
    [HAOI2011]Problem b 题解
  • 原文地址:https://www.cnblogs.com/wanstack/p/8872811.html
Copyright © 2011-2022 走看看