RabbitMQ 三种Exchange

RabbitMQ是一个消息代理，核心原理：发送消息，接收消息。

RabbitMQ主要用于组件之间的解耦，消息发送者无需知道消息使用者的存在，反之亦然。

                                单向解耦                                        双向解耦（如：RPC）

例如一个日志系统，很容易使用RabbitMQ简化工做量，一个Consumer进行消息的正常处理，另外一个Consumer复制对消息进行日志记录，只要在程序中指定两个Consumer所监听的queue以相同的方式绑定到同一个exchange便可，剩下的消息分发工做由RabbitMQ完成。

安装rabbitmq，请参考官网。

首先经过一个很是简单的”hello world“例子介绍如何使用RabbitMQ，而后再介绍其涉及的基本概念并对交换机和队列多作点介绍。

1、helloworld例子

本例很是简单——发送一个消息”hello world“，而后获取它并输出到屏幕。

总共须要两个程序，一个发送消息叫send.py，一个接受消息并打印消息内容叫receive.py。

该图为helloworld例子的原理图：生产者send.py（Productor）把消息（”hello world“）发送到一个名为”queue“的队列中，消费者receive.py从这个队列中获取消息。接下来看代码：

send.py(发送消息)

#!/usr/bin/env python
import pika
#第一步，链接RabbitMq服务器
rabbit_username='xxx' 
rabbit_password='xxx'
credentials = pika.PlainCredentials(rabbit_username, rabbit_password) 
connection = pika.BlockingConnection(pika.ConnectionParameters( host='x.x.x.x',credentials=credentials))
#channel是进行消息读写的通道
channel = connection.channel()

#第二步，建立一个名为queue的队列，而后把消息发送到这个队列

channel.queue_declare(queue='queue')

#第三步，如今能够发送消息，可是RabbitMQ不能把消息直接发送到队列，要发送到交换器，这个稍后介绍，这里使用默认交换器（exchange）,它使用一个空字符串标
#识，routing_key参数必须指定为队列名称，这里为queue

channel.basic_publish(exchange='',
                      routing_key='queue',
                      body='hello world')
print "send.py:send message 'hello world',wait for receive.py deal with this message"

#退出程序前，经过关闭链接保证消息已经投递到RabbitMq                                                                                               
connection.close()

 

 

receive.py（获取数据）

 

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

#!/usr/bin/env python
import pika

#第一步，一样链接RabbitMq服务器
rabbit_username='xxx'
rabbit_password='xxx'
credentials = pika.PlainCredentials(rabbit_username, rabbit_password)
connection = pika.BlockingConnection(pika.ConnectionParameters( host='x.x.x.x',credentials=credentials))
channel = connection.channel()

#为确保队列存在，再次执行queue_declare建立一个队列，咱们能够屡次运行该命令，可是只要一个队列会建立
#由于不能保证send.py先执行仍是receive.py先执行，因此重复声明队列来确保其存在
channel.queue_declare(queue='hellolxy')


#第三步，定义一个回调函数，当得到消息时，Pika库调用这个回调函数来处理消息，该回调函数将消息内容打印到屏幕
def callback(ch, method, properties, body):
    print "receive.py: Received message %r" % (body,)

#第四步，告诉rabbbitMq回调函数将从queue队列接收消息
channel.basic_consume(callback,
                      queue='queue',
                      no_ack=True)

#第五步，输入一个无限循环来等待消息数据并运行回调函数
print ' [*] Waiting for messages. To exit press CTRL+C'
channel.start_consuming()

 

如今在终端运行程序。首先，用send.py发送一条消息：

$ python send.py 
send.py:send message 'hello world',wait for receive.py deal with this message

生产者(producer)程序send.py每次运行后就会中止。如今运行receive.py来接收消息：

$ python receive.py 
receive.py: Received message 'hello world'  [*] Waiting for messages. To exit press CTRL+C

成功了！如今已经经过RabbitMQ发送了第一条消息。可是receive.py程序并无退出，它一直在准备获取消息，能够经过ctrl-c来中断它。

2、RabbitMQ基本概念

总结一下发送接收消息的过程：

经过上面例子对RabbitMQ有一个感性认识后，如今来介绍RabbitMQ中的基本概念。

Broker:消息队列服务器实体

消息：每一个消息都有一个路由键(routing key)的属性。就是一个简单的字符串。

connection：应用程序与broker的网络链接。

channel:几乎全部的操做都在channel中进行，channel是进行消息读写的通道。客户端可创建多个channel，每一个channel表明一个会话任务。

交换机：接收消息，根据路由键转发消息到绑定的队列。

绑定：一个绑定就是基于路由键将交换机和队列链接起来的路由规则，因此交换机不过就是一个由绑定构成的路由表。

举例：一个具备路由键“key1”的消息要发送到两个队列，queueA和queueB。要作到这点就要创建两个绑定，每一个绑定链接一个交换机和一个队列。二者都是由路由键“key1”触发，这种状况，交换机会复制一份消息并把它们分别发送到两个队列中。

队列：消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者链接到这个队列将其取走。

3、交换机

交换机用来接收消息，转发消息到绑定的队列，是rabbitMq中的核心。

交换机共有4种类型：direct，topic，headers和fanout。

为何不建立一种交换机来处理全部类型的路由规则？由于每种规则匹配时的CPU开销是不一样的，因此根据不一样需求选择合适交换机。

举例：一个"topic"类型的交换机会将消息的路由键与相似“dog.*”的模式进行匹配。一个“direct”类型的交换机会将路由键与 “dogs”进行比较。匹配末端通配符比直接比较消耗更多的cpu,因此若是用不到“topic”类型交换机带来的灵活性，就经过“direct”类型交 换机得到更高的处理效率。

一、Direct交换机：转发消息到routingKey指定队列（彻底匹配，单播）。

routingKey与队列名彻底匹配，若是一个队列绑定到交换机要求路由键为“dog”，则只转发routingkey标记为dog的消息，不会转发dog.puppy，也不会转发dog.guard等。

二、Topic交换机：按规则转发消息（最灵活，组播）

 Topic类型交换机经过模式匹配分配消息的routing-key属性。将路由键和某个模式进行匹配，此时队列须要绑定到一个模式上。

它将routing-key和binding-key的字符串切分红单词。这些单词之间用点隔开。它一样也会识别两个通配符：符号“#”和符号“*”。#匹配0个或多个单词，*匹配很少很多一个单词。

例如，binding key:*.stock.#匹配routing key: usd.stock和eur.stock.db，可是不匹配stock.nana。

例如，“audit.#”可以匹配到“audit.irs.corporate”，可是“audit.*”只会匹配到“audit.irs”。

三、Fanout交换机：转发消息到全部绑定队列（最快，广播）

fanout交换机不处理路由键，简单的将队列绑定到交换机上，每一个发送到交换机的消息都会被转发到与该交换机绑定的全部队列上。

很像子网广播，每台子网内的主机都得到了一份复制的消息。Fanout交换机转发消息是最快的。

四、Note

• 若是没有队列绑定在交换机上，则发送到该交换机上的消息会丢失。
• 一个交换机能够绑定多个队列，一个队列能够被多个交换机绑定。
• 还有一些其余类型的交换机类型，如header、failover、system等，如今在当前的RabbitMQ版本中均未实现。
• 由于交换机是命名实体，声明一个已经存在的交换机，可是试图赋予不一样类型是会致使错误。客户端须要删除这个已经存在的交换机，而后从新声明而且赋予新的类型。
• 交换机的属性：
  • 持久性：若是启用，交换机将会在server重启前都有效。
  • 自动删除：若是启用，那么交换机将会在其绑定的队列都被删掉以后删除自身。
  • 惰性:若是没有声明交换机，那么在执行到使用的时候会致使异常，并不会主动声明。

4、队列

• 队列的属性：
  • 持久性：若是启用，队列将在Server服务重启前都有效。
  • 自动删除：若是启用，那么队列将会在全部的消费者中止使用以后自动删除自身。
  • 惰性：若是没有声明队列，那么在执行到使用的时候会致使异常，并不会主动声明。
  • 排他性：若是启用，队列只能被声明它的消费者使用。        

 深刻了解可参考：