RabbitMQ（三） -- Publish/Subscribe-阿里云开发者社区

RabbitMQ（三） -- Publish/Subscribe

2017-11-29 1212

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介：

`rabbitmq`支持一对多的模式，一般称为发布/订阅。也就是说，生产者产生一条消息后，`rabbitmq`会把该消息分发给所有的消费者。

Exchanges

之前的教程中，仅仅使用了基本的消息模型：

生产者产生消息
把消息添加到消息队列
消费者接收消息

而在`rabbitmq完整的消息模型`中，并不是这样的。事实上，生产者并不知道消息是否发送到队列，而是把消息直接发送给`Exchanges`。

`Exchanges`的功能理解起来非常简单，它只负责接收生产者发送的数据并把这些数据添加到消息队列。但是，在存在多个消息队列的情况下，`Exchanges`必须知道每条消息要添加到哪一个消息队列。

`rabbitmq`提供了几种`Exchanges`，包括:`direct`, `topic`, `headers` and `fanout`。

这里，仅仅介绍fanout的使用。

channel.exchange_declare(exchange='news', type='fanout')

那么，发布消息：

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

Temporary queues

由于在生产者和消费者中需要指定相同的消息队列才能实现消息通信，那么如果不特殊指定某个消息队列会如何呢？
那么需要使用默认参数让系统给生成一个特定的消息队列。

result = channel.queue_declare()

Bindings

为了发送指定发送的消息队列，必须创建exchange和消息队列之间的关系：

channel.queue_bind(exchange='news', queue=result.method.queue)

例子

作为生产者的publish：

#!/usr/bin/env python
# coding=utf-8

import pika
import sys

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

channel.exchange_declare(exchange='news',
type='fanout')
for i in range(100):
    message = str(i) + 'Hello World!'
    channel.basic_publish(exchange='news', routing_key='', body=message)
    print " [x] Sent %r" % (message,)
    import time
    time.sleep(2)
connection.close()

作为消费者的subscribe：

#!/usr/bin/env python
# coding=utf-8
import pika

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

channel.exchange_declare(exchange='news', type='fanout')

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

channel.queue_bind(exchange='news',
queue=queue_name)

print ' [*] Waiting for news. 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()