<===
RabbitMQ消息队列(三):任务分发机制
上篇文章中,我们把每个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中。它的功能也很简单:从Producer接收Message,然后投递到queue中。Exchange需要知道如何处理Message,是把它放到那个queue中,还是放到多个queue中?这个rule是通过Exchange 的类型定义的。
我们知道有三种类型的Exchange:direct, topic 和fanout。fanout就是广播模式,会将所有的Message都放到它所知道的queue中。创建一个名字为logs,类型为fanout的Exchange:
channel.exchange_declare(exchange='logs',
type='fanout')
Listing exchanges
通过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)
2. Temporary queues
截至现在,我们用的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:
$ 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.
这个结果还是很好理解的。
尊重原创,转载请注明出处 anzhsoft: http://blog.csdn.net/anzhsoft/article/details/19617305
参考资料:
1. http://www.rabbitmq.com/tutorials/tutorial-three-python.html
相关推荐
RabbitMQ 是使用Erlang编写的一个开源的消息队列,本身支持很多的协议:AMQP,XMPP, SMTP, STOMP,也正是如此,使的它变的非常重量级,更适合于企业级的开发。同时实现了一个经纪人(Broker)构架,这意味着消息在发送...
在本篇教程中,我们要做的跟之前完全不一样 —— 分发一个消息给多个消费者(consumers)。这种模式被称为” 发布 / 订阅”。 交换器(Exchanges) 在本教程的前面部分,我们发送和接收到队列中的消息,现在是时候在...
【课程目录】:---第一章:RabbitMQ介绍----1-什么是消息中间件.mp4----2-RabbitMQ消息队列安装:window环境.mp4----3-RabbitMQ消息队列安装 :Linux环境.mp4----4-Rabbitmq入口示例:server.mp4----5-rabbitmq入口...
golang 推送系统 单机 客户端通过websocket连接到服务端 ...服务端从rabbitmq接收消息,根据消息所属通过websocket推送到具体用户 服务端从http api接收消息,根据消息所属通过websocket推送到具体用户
RabbitMQ实战: 高效部署分布式消息队列,高质量文档分享,请珍惜!
rabbitMQ官方Demo,包含01HelloWorld,02WorkQueues,03PublishSubscribe,04Routing,05Topics,06RPC六个示例
采用python编写的批量删除rabbitmq的队列或交换机。 1.修改rabbitmq_delete.py中rabbitmq的配置; 2.执行以下命令: 删除队列: python3 rabbitmq_delete.py -k ‘udata.climb’ -d 1 删除交换机: python3 rabbitmq_...
文章目录rabbitmq7种实现方式搭建maven项目引入依赖创建连接简单队列消息生产者消息消费者work queues 工作队列生产者消费者能者多劳(公平分发):消费能力强则消费更多消息Publish/Subscribe 发布订阅模式生产者...
RabbitMQ实战 高效部署分布式消息队列 RabbitMQ实战 高效部署分布式消息队列
springcloud bus rabbitmq 分布式队列 http://knight-black-bob.iteye.com/blog/2356839
基于rabbitmq实现的消息队列组件,附所有源代码,大家可以放心的使用,可用于商业程序,欢迎大家交流。
将消息从 RabbitMQ 队列转储到文件,而不影响队列。 安装 下载一个版本 可以在页面上找到预编译的二进制包。 从源代码编译 如果您安装了 ,则可以通过运行以下命令从源代码安装 rabbitmq-dump-queue: go get ...
导入里面的sql,也就是一个user表,直接运行,访问...可以在消息适配器MessageListenerAdapter里配置队列和方法的映射queueOrTagToMethodName,指定哪个队列调用消费者哪个方法,同步的需要有返回值
基于RabbitMQ消息队列的路由分发实例源码,直接可以运行。
SpringBoot+RabbitMQ,springboot与rabbitmq消息队列的整合。
rabbitMQ 死信队列 完整例子
Erlang和RabbitMQ 消息队列
RabbitMQ是由erlang语言开发,基于AMQP(Advanced Message Queue 高级消息队列协议)协议实现的消息队列,它是一种应用程序之间的通信方法,消息队列在分布式系统开发中应用非常广泛。 MQ 为Message Queue , 消息...
最新stomp.js源码,连接rabbitmq等mq使用: // 初始化 ws 对象 if (location.search == '?ws') { var ws = new WebSocket('ws://localhost:15674/ws'); } else { var ws = new SockJS('...