什么是RabbitMQ
在了解什么是RabbitMQ之前有必要先说说amqp
AMQP
AMQP,即Advanced Message Queuing Protocol,高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。消息中间件主要用于组件之间的解耦,消息的发送者无需知道消息使用者的存在,反之亦然。 它可以使对应的客户端(client)与对应的消息中间件(broker)进行交互。消息中间件从发布者(publisher)那里收到消息(发布消息的应用,也称为producer),然后将他们转发给消费者(consumers,处理消息的应用)。由于AMQP是一个网络协议,所以发布者、消费者以及消息中间件可以部署到不同的物理机器上面。在 2006 年的 6 月,Cisco 、Redhat、iMatix 等联合制定了 AMQP 的公开标准,
RabbitMQ
RabbitMQ是一个开源的AMQP实现,服务器端用Erlang语言编写,支持多种客户端,如:Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等,支持AJAX。用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗。AMQP的实现还有ActiveMQ、Kafka、RocketMQ等。RabbitMQ是由RabbitMQ Technologies Ltd开发并且提供商业支持的。该公司在2010年4月被SpringSource(VMWare的一个部门)收购。在2013年5月被并入Pivotal。
AMQP的应用场景
我们假设你需要实现一个用户注册功能,注册成功之后需要向用户发送邮件、发送短信、存日志记录。你很可能会采用如下方式
但是渐渐的你发现,注册功能好像太慢了,毕竟一个注册要调用发短信系统,发邮件系统,存日志系统,这些都是很耗时的,你在想,我能不能异步去做这些事情呢?于是你采用了如下的方案
现在注册功能快多了,你觉得很开心,但是随着公司的用户数提升,你开始渐渐的发现问题了。一个注册功能就要启动三个线程,而我们知道一台服务器的线程数是有上限的,线程上限之后,后续的用户注册只能等待。你们领导让你优化这个问题,你找啊找,于是AMQP出现了,你采用了如下方式
你把一个用户注册的消息放到MQ队列中,其他消费者从这个队列里面去取出来然后实现自己的逻辑,完美。
AMQP解决的问题
- 信息的发送者和接收者如何维持这个连接,如果一方的连接中断,这期间的数据如何方式丢失?
- 如何降低发送者和接收者的耦合度?
- 如何让Priority高的接收者先接到数据?
- 如何做到Load balance?有效均衡接收者的负载?
- 如何有效的将数据发送到相关的接收者?也就是说将接收者subscribe 不同的数据,如何做有效的filter。
- 如何做到可扩展,甚至将这个通信模块发到cluster上?
- 如何保证接收者接收到了完整,正确的数据?
AMQP的实现之RabbitMQ
应用场景架构
- Client A , B:也叫Producer,数据(Message)的发送方。一个Message有两个部分:Payload(有效载荷)和Label(标签)。Payload顾名思义就是传输的数据,Label是Exchange的名字或者说是一个tag,它描述了payload,而且RabbitMQ也是通过这个label来决定把这个Message发给哪个Consumer。AMQP仅仅描述了label,而RabbitMQ决定了如何使用这个label的规则。
- RabbitMQ Server:也叫broker server,他的角色就是维护一条从Producer到Consumer的路线,保证数据能够按照指定的方式进行传输。
- Client 1,2,3:也叫Consumer,数据的接收方。把queue比作是一个有名字的邮箱。当有Message到达某个邮箱后,RabbitMQ把它发送给它的某个订阅者即Consumer。当然可能会把同一个Message发送给很多的Consumer。在这个Message中,只有payload,label已经被删掉了。对于Consumer来说,它是不知道谁发送的这个信息的。AMQP协议本身就不支持。如果需要Producer的信息可以考虑把Producer的信息包含在其说发送的Message中
RabbitMQ Server
为了保证数据从Producer到Consumer的正确传递,AMQP中对于Server层定义了几个概念
- exchanges:消息交换机,它指定消息按什么规则,路由到哪个队列
- queues:消息队列,每个消息都会被投入到一个或多个队列
- bindings:它的作用就是把exchange和queue按照路由规则绑定起来
- Routing Key:路由关键字,exchange根据这个关键字进行消息投递
还有几个概念是上述图中没有标明的
- Connection:就是一个TCP的连接。Producer和Consumer都是通过TCP连接到RabbitMQ Server的。以后我们可以看到,程序的起始处就是建立这个TCP连接。
- Channel:虚拟连接。它建立在上述的TCP连接中。数据流动都是在Channel中进行的。也就是说,一般情况是程序起始建立TCP连接,第二步就是建立这个Channel。
- Vhost:虚拟主机,一个broker里可以开设多个vhost,用作不同用户的权限分离。每个virtual host本质上都是一个RabbitMQ Server,拥有它自己的queue,exchagne,和bings rule等等。这保证了你可以在多个不同的application中使用RabbitMQ。消息队列执行过程
- 客户端连接到消息队列服务器,打开一个Channel。
- 客户端声明一个Exchange,并设置相关属性。
- 客户端声明一个Queue,并设置相关属性。
- 客户端使用Routing key,在Exchange和Queue之间建立好绑定关系。
- 客户端投递消息到Exchange。
- Exchanges接收到消息后,就根据消息的key和已经设置的Binding,进行消息路由,将消息投递到一个或多个队列里。
而对于Exchanges来说,他的路由规则有以下三种
- Direct exchange:完全根据key进行投递的叫做Direct交换机。如果Routing key匹配, 那么Message就会被传递到相应的queue中。其实在queue创建时,它会自动的以queue的名字作为routing key来绑定那个exchange。例如,绑定时设置了Routing key为”abc”,那么客户端提交的消息,只有设置了key为”abc”的才会投递到队列。
- Fanout exchange:不需要key的叫做Fanout交换机。它采取广播模式,一个消息进来时,投递到与该交换机绑定的所有队列。
- Topic exchange:对key进行模式匹配后进行投递的叫做Topic交换机。比如符号”#”匹配一个或多个词,符号””匹配正好一个词。例如”abc.#”匹配”abc.def.ghi”,”abc.”只匹配”abc.def”。
放在最后
到此为止,RabbitMQ的基本概念讲完了,但是这边还有一些问题,比如如果突发宕机,RabbitMQ如何能够保证消息不丢失呢?我们知道,任何一个组件都是有理论性能上限的,如果数据量大到RabbitMQ也处理不过来那该怎么办呢?在下一章中我们一起来分析这些问题。