RabbitMQ的介绍

发表时间:2017-11-10 09:38:31 浏览量( 28 ) 留言数( 0 )

学习目标:

1、了解RabbitMQ的发展历史。

2、了解RabbitMQ的优势


学习过程:

一、RabbitMQ的简介

官方网址:https://www.rabbitmq.com/   

    RabbitMQ是最专业的MQ了,支持AMQP标准协议。AMQP即Advanced Message Queuing Protocol,高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。消息中间件主要用于组件之间的解耦,消息的发送者无需知道消息使用者的存在,反之亦然。

    AMQP的主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。

    RabbitMQ是一个开源的AMQP实现,服务器端用Erlang语言编写,支持多种客户端,如:Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等,支持AJAX。用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗。

   虽然效率上在某些场景上可能比不上其他MQ。但是在易用性,稳定性安全性等各方面对比其他的MQ技术都是胜出一筹的,RabbitMQ相对其他的MQ技术主要有以下的优点:

1、具有高可用高并发的优点,适合集群服务器。

2、健壮、稳定、易用、跨平台、支持多种语言、文档齐全,对比ActiveMQ只对Java语言的支持,RabbitMQ支持AMQP的协议,可以支持大部分的语言。

3、使用消息确认机制和持久化机制,可靠性高,如果客户端处理异常,可以重发消息。RabbitMQ是工业级别的,大量金融机构使用,足见其可靠性高。

    RabbitMQ最大的优势在于它提供了最灵活的消息路由策略、高可用和可靠性,可靠性又分为两部分(消息可靠性和软件可靠性),以及丰富的插件、平台支持和完善的文档。然而,由于AMQP协议本身的灵活性导致了它比较重量,所以造成了它相比某些队列服务(如Kafka)吞吐量处于下风。因此,当选择一个消息队列服务时,关键还是看需求上更看重消息吞吐量、消息堆积能力还是路由灵活性、高可用、可靠传输


二、概念

1、基本概念说明:

producer:消息生产者,就是投递消息的程序。

consumer:消息消费者,就是接受消息的程序。

Connection:连接,对于RabbitMQ而言,其实就是一个位于客户端和Broker之间的TCP连接。

ConnectionFactory:为Connection的制造工厂。

channel:消息通道,在客户端的每个连接里,可建立多个channel,每个channel代表一个会话任务。仅仅创建了客户端到Broker之间的连接后,客户端还是不能发送消息的。需要为每一个Connection创建Channel,AMQP协议规定只有通过Channel才能执行AMQP的命令。一个Connection可以包含多个Channel。之所以需要Channel,是因为TCP连接的建立和释放都是十分昂贵的,如果一个客户端每一个线程都需要与Broker交互,如果每一个线程都建立一个TCP连接,暂且不考虑TCP连接是否浪费,就算操作系统也无法承受每秒建立如此多的TCP连接。RabbitMQ建议客户端线程之间不要共用Channel,至少要保证共用Channel的线程发送消息必须是串行的,但是建议尽量共用Connection。

Message: 由Header和Body组成,Header是由生产者添加的各种属性的集合,包括Message是否被持久化、由哪个Message Queue接

Broker:简单来说就是消息队列服务器实体。

Exchange:接受生产者发送的消息,并根据Binding规则将消息路由给服务器中的队列。ExchangeType决定了Exchange路由消息的行

Queue:消息队列载体,每个消息都会被投入到一个或多个队列。用于存储还未被消费者消费的消息。

Binding:绑定,它的作用就是把exchange和queue按照路由规则绑定起来。

Routing Key:路由关键字,exchange根据这个关键字进行消息投递。

vhost:虚拟主机,一个broker里可以开设多个vhost,用作不同用户的权限分离。


2、消息队列的使用过程大概如下:

(1)客户端连接到消息队列服务器,打开一个channel。

(2)客户端声明一个exchange,并设置相关属性。

(3)客户端声明一个queue,并设置相关属性。

(4)客户端使用routing key,在exchange和queue之间建立好绑定关系。

(5)客户端投递消息到exchange。

    exchange接收到消息后,就根据消息的key和已经设置的binding,进行消息路由,将消息投递到一个或多个队列里。

    exchange也有几个类型,RabbitMQ常用的Exchange Type有fanout、direct、topic、headers这四种(AMQP规范里还提到两种Exchange Type,分别为system与自定义,这里不予以描述),以后我们会详细接收完全根据key进行投递的叫做Direct交换机,例如,绑定时设置了routing key为”abc”,那么客户端提交的消息,只有设置了key为”abc”的才会投递到队列。对key进行模式匹配后进行投递的叫做Topic交换机,符号”#”匹配一个或多个词,符号”*”匹配正好一个词。例如”abc.#”匹配”abc.def.ghi”,”abc.*”只匹配”abc.def”。还有一种不需要key的,叫做Fanout交换机,它采取广播模式,一个消息进来时,投递到与该交换机绑定的所有队列。

3、消息确认Message acknowledgment

    和ActiveMQ类似。在实际应用中,可能会发生消费者收到Queue中的消息,但没有处理完成就宕机(或出现其他意外)的情况,这种情况下就可能会导致消息丢失。为了避免这种情况发生,我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ,RabbitMQ收到消息回执(Message acknowledgment)后才将该消息从Queue中移除;如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开,则RabbitMQ会将该消息发送给其他消费者(如果存在多个消费者)进行处理。这里不存在timeout概念,一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者,除非它的RabbitMQ连接断开。

    这里会产生另外一个问题,如果我们的开发人员在处理完业务逻辑后,忘记发送回执给RabbitMQ,这将会导致严重的bug——Queue中堆积的消息会越来越多;消费者重启后会重复消费这些消息并重复执行业务逻辑…另外pub message是没有ack的。

4、消息持久化Message durability

RabbitMQ支持消息的持久化,也就是数据写在磁盘上,为了数据安全考虑,我想大多数用户都会选择持久化。如果我们希望即使在RabbitMQ服务重启的情况下,也不会丢失消息,我们可以将Queue与Message都设置为可持久化的(durable),这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。但依然解决不了小概率丢失事件的发生(比如RabbitMQ服务器已经接收到生产者的消息,但还没来得及持久化该消息时RabbitMQ服务器就断电了),如果我们需要对这种小概率事件也要管理起来,那么我们要用到事务。由于这里仅为RabbitMQ的简单介绍,所以这里将不讲解RabbitMQ相关的事务。消息队列持久化包括3个部分:

  (1)exchange持久化,在声明时指定durable => 1

  (2)queue持久化,在声明时指定durable => 1

  (3)消息持久化,在投递时指定delivery_mode => 2(1是非持久化)

    如果exchange和queue都是持久化的,那么它们之间的binding也是持久化的。如果exchange和queue两者之间有一个持久化,一个非持久化,就不允许建立绑定。

5、处理数量Prefetch count

    前面我们讲到如果有多个消费者同时订阅同一个Queue中的消息,Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同,就有可能会导致某些消费者一直在忙,而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数,比如我们设置prefetchCount=1,则Queue每次给每个消费者发送一条消息;消费者处理完这条消息后Queue会再给该消费者发送一条消息。

6、绑定binding key

   在绑定(Binding)Exchange与Queue的同时,一般会指定一个binding key;消费者将消息发送给Exchange时,一般会指定一个routing key;当binding key与routing key相匹配时,消息将会被路由到对应的Queue中。这个将在Exchange Types章节会列举实际的例子加以说明。

在绑定多个Queue到同一个Exchange的时候,这些Binding允许使用相同的binding key。

binding key 并不是在所有情况下都生效,它依赖于Exchange Type,比如fanout类型的Exchange就会无视binding key,而是将消息路由到所有绑定到该Exchange的Queue