转载

RabbitMQ消息队列（五）：Routing 消息路由

上篇文章中，我们构建了一个简单的日志系统。接下来，我们将丰富它：能够使用不同的severity来监听不同等级的log。比如我们希望只有error的log才保存到磁盘上。

1. Bindings绑定

上篇文章中我们是这么做的绑定：

channel.queue_bind(exchange=exchange_name,
                   queue=queue_name)

绑定其实就是关联了exchange和queue。或者这么说：queue对exchagne的内容感兴趣，exchange要把它的Message deliver到queue中。实际上，绑定可以带routing_key 这个参数。其实这个参数的名称和basic_publish 的参数名是相同了。为了避免混淆，我们把它成为binding key。使用一个key来创建binding :

channel.queue_bind(exchange=exchange_name,
                   queue=queue_name,
                   routing_key='black')

对于fanout的exchange来说，这个参数是被忽略的。

2. Direct exchange

Direct exchange的路由算法非常简单：通过binding key的完全匹配，可以通过下图来说明。

exchange X和两个queue绑定在一起。Q1的binding key是orange。Q2的binding key是black和green。当P publish key是orange时，exchange会把它放到Q1。如果是black或者green那么就会到Q2。其余的Message都会被丢弃。

3. Multiple bindings

多个queue绑定同一个key是可以的。对于下图的例子，Q1和Q2都绑定了black。也就是说，对于routing key是black的Message，会被deliver到Q1和Q2。其余的Message都会被丢弃。 RabbitMQ消息队列（五）：Routing 消息路由

4. Emitting logs

首先是我们要创建一个direct的exchange：

channel.exchange_declare(exchange='direct_logs',
                         type='direct')

我们将使用log的severity作为routing key，这样Consumer可以针对不同severity的log进行不同的处理。 publish：

channel.basic_publish(exchange='direct_logs',
                      routing_key=severity,
                      body=message)

我们使用三种severity：'info', 'warning', 'error'.

5. Subscribing

对于queue，我们需要绑定severity：

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

for severity in severities:
    channel.queue_bind(exchange='direct_logs',
                       queue=queue_name,
                       routing_key=severity)

6. 最终版本

The code for emit_log_direct.py:

#!/usr/bin/env python
import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs',
                         type='direct')

severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='direct_logs',
                      routing_key=severity,
                      body=message)
print " [x] Sent %r:%r" % (severity, message)
connection.close()

The code for receive_logs_direct.py:

#!/usr/bin/env python
import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs',
                         type='direct')

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

severities = sys.argv[1:]
if not severities:
    print >> sys.stderr, "Usage: %s [info] [warning] [error]" % /
                         (sys.argv[0],)
    sys.exit(1)

for severity in severities:
    channel.queue_bind(exchange='direct_logs',
                       queue=queue_name,
                       routing_key=severity)

print ' [*] Waiting for logs. To exit press CTRL+C'

def callback(ch, method, properties, body):
    print " [x] %r:%r" % (method.routing_key, body,)

channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)

channel.start_consuming()

我们想把warning和error的log记录到一个文件中：

$ python receive_logs_direct.py warning error > logs_from_rabbit.log

打印所有log到屏幕：

$ python receive_logs_direct.py info warning error
 [*] Waiting for logs. To exit press CTRL+C

参考资料： 1. http://www.rabbitmq.com/tutorials/tutorial-four-python.html 2. http://blog.csdn.net/anzhsoft/article/details/19630147

正文到此结束

所属分类： rabittmq教程

本文标签：
版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
本文海报： 生成海报一生成海报二

其他链接

关于本站

本站定位：个人技术类博客

本站作用：写博客、记日志、闲聊扯淡鼓捣技术。

问题交流

RabbitMQ消息队列（五）：Routing 消息路由

1. Bindings绑定

2. Direct exchange

3. Multiple bindings

4. Emitting logs

5. Subscribing

6. 最终版本

热门推荐

相关文章

rabittmq教程

说给你听

本文目录

随机标签

书籍教程

近期评论

网站信息

其他链接

关于本站

问题交流