python 目录
(一)、交换 (Exchanges)
-- 1.1 武sir 经典 Exchanges 案例展示。
(二)、临时队列( Temporary queues )
(三)、绑定(Bindings)
(四)、汇总(Putting it all together)
python系列之 RabbitMQ -- Publish/Subscribe 【发布和订阅】
>>前面的部分我们创建了一个工作队列(work queue). 设想是每个任务都能分发到一个 worker【queue】,这一部分我们将会做一些完全不同的事情
-- 我们将会分发一个消息到多个消费方(consumer),这种模式被誉为 发布/订阅(publish/subscribe)模式
>>为了阐明这种模式,我们将要创建一个简单的日志系统。
由两部分程序组成 --
第一部分将要发布日志消息,第二部分接收并打印
在我们的日志系统中每个接收程序(receiver)将接收消息并复制消息内容,这样我们将会运行一个receiver 记录日志到磁盘;与此同时我们运行另一个receiver输入日志到屏幕查看。
本质上,发布日志消息将会广播到所有的 receivers。
前奏:交换 (Exchanges) 【武SIR前-序曲】
exchange类型可用: direct , topic , headers 和 fanout 。
我们将要对最后一种进行讲解 --- fanout
一、消息发布订阅
发布订阅和简单的消息队列区别在于,发布订阅会将消息发送给所有的订阅者,而消息队列中的数据被消费一次便消失。所以,RabbitMQ实现发布和订阅时,会为每一个订阅者创建一个队列,而发布者发布消息时,会将消息放置在所有相关队列中。
exchange type = fanout
1、发布者生产者
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()
2、订阅者消费者。
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()
(一)、交换 (Exchanges)
在前面的部分我们从一个队列来回发送并接收消息。现在介绍Rabbit中的完整消息模式。
让我们迅速回顾一下前面的章节的内容:
>> 一个 Producer 是一个发送消息的用户程序
>> 一个 queue 是一个存储消息的缓冲区
>> 一个 Consumer 是一个接收消息的用户程序
RabbitMQ的消息模式的核心思想是:
生产者(Producer) 从不将消息直接发送到一个队列(queue)中,实际上,很多时候生产者甚至不知道一个消息是否要分发到所有队列.换言之,生产者(producer)只能够发送消息到一个交换区 Exchange.. 对exchange发送消息是 -->分发到所有队列照中.从生产者producer方接收消息,从另一边将消息push到队列中。exchange必须清楚知道接收到的消息要如何处理. 是要将消息发送到一个指定queue? 是要将消息发送到多个queue? 还是丢弃?这个规则需要通过 exchange type 来定义。
![](https://images2015.cnblogs.com/blog/891620/201607/891620-20160730091720450-2015242282.png)
这里有几种exchange类型可用: direct , topic , headers 和 fanout 。
我们将要对最后一种进行讲解 --- fanout 。 我们创建一个 这种类型的exchange 并命名为logs:
channel.exchange_declare(exchange='logs',type='fanout')
关于Exchange 的几种模式:
http://blog.csdn.net/songfreeman/article/details/50953288exchanges列表
使用rabbitmqclt管理工具显示服务器上的exchanges列表
$ sudo rabbitmqctl list_exchanges Listing exchanges ... logs fanout amq.direct direct amq.topic topic amq.fanout fanout amq.headers headers ...done.
在这个列表中有一些amq.* exchange和默认的exchange,这些都是默认创建的,但是这些未必是你所需要的。
匿名的exchange
在前面部分我们知道空的exchange, 但仍然能够发送消息到队列中,只是因为我们使用的是我们定义的空字符串“ ”exchange(默认的exchange)
回忆一下我们之前怎么发布一个消息:
channel.basic_publish(exchange='',routing_key='hello', #指定哪个队列。 body=message)
这个exchange参数就是这个exchange的名字. 空字符串标识默认的或者匿名的exchange:如果存在routing_key, 消息路由到routing_key指定的队列中。
现在我们可能够发布消息到我们自己命名的exchange:
channel.basic_publish(exchange='logs', routing_key='', body=message)
(二)、临时队列( Temporary queues )
你应该记得我们之前使用有一个特定名字的队列( hello、task_queue). 设置队列名对我们来说是至关重要的 --- 我们需要给消费方指定同样的队列名字。 要在生产者和消费者之间共享队列,给队列设置一个名字是非常重要的。但是这不是我们日志应用的关键,我们希望获取到所有的日志消息,而不是他们的一个子集。我们只对当前活动的消息敢兴趣,对已经发过的旧的消息不关心。为了解决这个问题我们需要做两件事:
首先:
无论什么时候我们连接到Rabbit我们需要一个空的、新的队列。为了实现这个我们可以创建个随机名的队列,
或者,
更好的-让服务端选择一个随机的队列名给我们,我们可以不给queue_declare方法设置 queue参数来实现。
result = channel.queue_declare()
- 这样, result.method.queue 包含一个随机的队列名, 比如:看起来像 amq.gen-JzTY20BRgKO-HjmUJj0wLg.
- 其次,一旦我们断开consumer连接,这个队列名将自动删除。这里有一个标识设置:
result = channel.queue_declare(exclusive=True)
(三)、绑定(Bindings)
![](https://images2015.cnblogs.com/blog/891620/201607/891620-20160730091721263-293607644.png)
我们已经创建了一个 fanout exchange 和一个队列queue, 现在我们需要告诉exchange发送消息到我们的queue中, 这个exchange与队列queue之间的关系我们成为 绑定(Bindings)
现在logs exchange 将要发送消息到我们的队列channel.queue_bind(exchange='logs', queue=result.method.queue) #固定的写法
你可以在Server端通过rabbitmqctl list_bindings命令查看绑定信息
(四)、汇总(Putting it all together)
![](https://images2015.cnblogs.com/blog/891620/201607/891620-20160730091722106-446928262.png)
生产者(Producer)程序发出log消息,和前面介绍的章节没什么太大的区别。
>> 主要的改变是我们想要发送消息到我们指定的logs exchange,而不是之前的那种匿名的exchange,那样的routing_key 还要指定发送的队列 如"myqueue",很死性的。
- 我们之前使用匿名exchange发送消息时Producer需要提供一个 routing_key,这样就直接绑定到__队列上,, 跳过了中间件的 exchange。
- 但当我们指定exchange为fanout exchange时,这个值(routing_key)将忽略。
下面是 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()
发送消息时,如果还没有队列绑定到这个exchange上的话,消息将会丢失。 但这个对我们来说是OK的;如果还没有消费者(consumer) 监听上我们可以安全的放弃这条消息。
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
使用 rabbitmqlctl list_bindings 你能验证代码确实创建了你想要的binding和队列。运行两个 receive_logs.py 程序你可以看到:
$ sudo rabbitmqctl list_bindings Listing bindings ... logs exchange amq.gen-JzTY20BRgKO-HjmUJj0wLg queue [] logs exchange amq.gen-vso0PVvyiRIL2WoV3i48Yg queue [] ...done.
这个结果的解释非常直白: 从 logs exchange 出来的数据发送服务端自动分配的到两个队列名中,这也是我们预期的。