标签：
>>前面的部分我们创建了一个工作队列（work queue）. 设想是每个任务都能分发到一个 worker【queue】，这一部分我们将会做一些完全不同的事情
-- 我们将会分发一个消息到多个消费方（consumer），这种模式被誉为 发布/订阅（publish/subscribe）模式
>>为了阐明这种模式,我们将要创建一个简单的日志系统。
由两部分程序组成 --
第一部分将要发布日志消息，第二部分接收并打印
在我们的日志系统中每个接收程序（receiver）将接收消息并复制消息内容，这样我们将会运行一个receiver 记录日志到磁盘；与此同时我们运行另一个receiver输入日志到屏幕查看。
本质上，发布日志消息将会广播到所有的 receivers。
exchange类型可用： direct , topic , headers 和 fanout 。
我们将要对最后一种进行讲解 --- fanout
发布订阅和简单的消息队列区别在于，发布订阅会将消息发送给所有的订阅者，而消息队列中的数据被消费一次便消失。所以，RabbitMQ实现发布和订阅时，会为每一个订阅者创建一个队列，而发布者发布消息时，会将消息放置在所有相关队列中。
exchange type = fanout
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()
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()
生产者(Producer) 从不将消息直接发送到一个队列(queue)中，实际上，很多时候生产者甚至不知道一个消息是否要分发到所有队列.换言之，生产者(producer)只能够发送消息到一个交换区 Exchange.. 对exchange发送消息是 -->分发到所有队列照中.从生产者producer方接收消息，从另一边将消息push到队列中。exchange必须清楚知道接收到的消息要如何处理. 是要将消息发送到一个指定queue？ 是要将消息发送到多个queue？ 还是丢弃？这个规则需要通过 exchange type 来定义。
channel.exchange_declare(exchange=‘logs‘,type=‘fanout‘)
$ sudo rabbitmqctl list_exchanges Listing exchanges ... logs fanout amq.direct direct amq.topic topic amq.fanout fanout amq.headers headers ...done.
channel.basic_publish(exchange=‘‘,routing_key=‘hello‘, ＃指定哪个队列。 body=message)
channel.basic_publish(exchange=‘logs‘, routing_key=‘‘, body=message)
result = channel.queue_declare()
result = channel.queue_declare(exclusive=True)
现在logs exchange 将要发送消息到我们的队列channel.queue_bind(exchange=‘logs‘, queue=result.method.queue) #固定的写法
你可以在Server端通过rabbitmqctl list_bindings命令查看绑定信息
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()
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()
$ python receive_logs.py > logs_from_rabbit.log
$ python receive_logs.py
$ python emit_log.py
$ sudo rabbitmqctl list_bindings Listing bindings ... logs exchange amq.gen-JzTY20BRgKO-HjmUJj0wLg queue [] logs exchange amq.gen-vso0PVvyiRIL2WoV3i48Yg queue [] ...done.
这个结果的解释非常直白： 从 logs exchange 出来的数据发送服务端自动分配的到两个队列名中，这也是我们预期的。
Part1.2 、RabbitMQ -- Publish/Subscribe 【发布和订阅】
标签：
原文地址：http://www.cnblogs.com/zhangju/p/5720228.html
