RabbitMQ学习笔记(四)

发布/订阅

ps: 使用pika python客户端
前面学习了搭建工作队列,每一个任务只是分发给了一个工作者。但是现在我想实现将一个消息分发给多个消费者。就是设计模式中的”观察者模式”–发布/订阅。

为了描述这种模式,构件一个简单的日志系统。包括两个程序–第一个程序负发送日志消息,第二个程序负责获取消息并输出内容。

在日志系统中,所有运行的接收方程序都会接收消息,我们让其中一个接收者将日志写入磁盘,另一个接收者将日志输出到屏幕上。日志消息是被广播到所有的接收者。

交换机####

分析前面的教程:

  • 发布者(producer)是发布消息的应用程序
  • 队列(queue)用于消息的存储
  • 消费者(consumer)是接受消息的应用程序。
    RabbitMQ其实不是直接将消息发送到队列中去,事实上,发布者是将消息交给excheng(交换机)。交换机一边从消息发布方接收消息,一边将消息推送到队列。交换机必须知道如何处理他所接收的消息,是应该推送到指定的队列还是多个队列,或者直接忽视消息。这些规则是有交换机的类型来决定。

《RabbitMQ学习笔记(四)》

RabbitMQ提供几种类型的交换机可供选择:

Itemname
直连交换机direct exchange
扇型交换机fanout exchange
主题交换机topic exchange
头交换机headers exchange

我们使用一下方法来创建一个扇型交换机

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

Fanout exchange将所有生产者发送到本交换机上的消息全部像风扇转动,将所有的消息发给所有的队列。

匿名交换机

前面我们没有提到减缓及,但是仍然能够将消息发送到队列中。因为我们使用了命名为空字符串的默认交换机。

channel.basic_publish(echange='',
  routing_key='hello',
  body=message)

exchange参数就是交换机的名字,空字符串表示匿名交换机,消息将发送到指定的routing_key指定的队列。
我们可以发送一个消息到一个具名的交换机

channel.basic_publish(exchange='logs',
  routing='',
  body=message)

在扇型交换机中,routing_key是不需要的。

临时队列

队列的名字可以由我们手动创建,但是也可以使用系统给我们创建一个随机的队列名字。只需要在创建队列的函数中加上参数就可以。

result = channel.queue_declare(exclusive=True)

这样就可以创建一个匿名队列(形式为amq.gen-*),我们可以通过result.method.queue来获取这个随机的队列名。当消费者断开连接的时候吗,这个队列就会被立即删除。

绑定(bindings)

《RabbitMQ学习笔记(四)》

我们创建一个扇型交换机和一个队列,现在需要告诉交换机如发送消息给我们的队列,交换机和队列之间的关系叫做绑定(binding)

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

现在,logs交换机将会把消息添加到我们的队列中。

《RabbitMQ学习笔记(四)》

Coding

emit_log.py

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()

我们创建了一个fanout类型的交换机,发送消息时将消息发送到这个交换机上。
receives_logs.py

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()
  • 消费者创建了一个fanout类型的交换机,这里重复创建是如果消费者程序先运行,不创建交换机,是不允许将消息发送到一个不存在的交换机的。
  • 消费者创建了一个匿名队列,然后将这个匿名队列和交换机进行bind。
  • 消费者等待从绑定的交换机队列中获取消息。
    打开两个终端一个终端将日志保存到日志文件;另一个将日志输出到屏幕上。
$ python receive_logs.py > logs_from_rabbit.log

这个终端运行的消费者将log信息打印出来

$ python receive_logs.py

发送日志信息:

$ python emit_log.py

使用rabbitmqctl list_bindings可以查看已经创建的队列绑定

$ sudo rabbitmqctl list_bindings
Listing bindings ...
 ...
logs    amq.gen-TJWkez28YpImbWdRKMa8sg==                []
logs    amq.gen-x0kymA4yPzAT6BoC/YP+zw==                []
...done.

显示我们创建的两个匿名队列都绑定到了fanout类型交换机logs上面。

待续。。。

参考文章http://rabbitmq.mr-ping.com/tutorials_with_python/[3]Publish_Subscribe.html

    原文作者:嘿嘿_小余同学
    原文地址: https://www.jianshu.com/p/6ad54d8742f6
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞