发布/订阅
ps: 使用pika python客户端
前面学习了搭建工作队列，每一个任务只是分发给了一个工作者。但是现在我想实现将一个消息分发给多个消费者。就是设计模式中的"观察者模式"--发布/订阅。
为了描述这种模式，构件一个简单的日志系统。包括两个程序--第一个程序负发送日志消息，第二个程序负责获取消息并输出内容。
在日志系统中，所有运行的接收方程序都会接收消息，我们让其中一个接收者将日志写入磁盘，另一个接收者将日志输出到屏幕上。日志消息是被广播到所有的接收者。
交换机####
分析前面的教程：
- 发布者(producer)是发布消息的应用程序
- 队列(queue)用于消息的存储
- 消费者(consumer)是接受消息的应用程序。
RabbitMQ其实不是直接将消息发送到队列中去，事实上，发布者是将消息交给excheng(交换机)。交换机一边从消息发布方接收消息，一边将消息推送到队列。交换机必须知道如何处理他所接收的消息，是应该推送到指定的队列还是多个队列，或者直接忽视消息。这些规则是有交换机的类型来决定。
RabbitMQ提供几种类型的交换机可供选择：
| Item | name |
|---|---|
| 直连交换机 | 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)
我们创建一个扇型交换机和一个队列，现在需要告诉交换机如发送消息给我们的队列，交换机和队列之间的关系叫做绑定(binding)
channel.queue_bind(exchange='logs',
queue=result.method.queue)
现在，logs交换机将会把消息添加到我们的队列中。
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
网友评论