Day10_04_消息队列之RabbitMQ应用场景及原理

我会带着你远行 2024-01-20 08:08 17阅读 0赞

## RabbitMQ应用场景及原理 ##

### 一. 概述 ###

RabbitMQ是一个由erlang开发的对AMQP(Advanved Message Queue Protocol)高级消息队列协议的开源实现.

### 二. RabbitMQ 应用场景 ###

#### 1. 异步处理 ####

场景说明: 用户注册后,需要发注册邮件和注册短信,传统的做法有两种--1️⃣.串行的方式;2️⃣.并行的方式.

1.1 串行方式实现

将注册信息写入数据库后,发送注册邮件,再发送注册短信,以上三个任务全部完成后才返回给客户端.这有一个问题是,邮件,短信并不是必须的,它只是一个通知,而这种做法却让客户端等待了没有必要等待的东西.  
![Image 1][]![20190612231605129.png][]

1.2 并行方式实现

将注册信息写入数据库后,发送邮件的同时,发送短信,以上三个任务完成后,返回给客户端,并行的方式能提高处理的时间.  
![Image 1][]![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70][]

假设三个业务节点分别使用50ms,串行方式使用时间150ms,并行使用时间100ms.虽然并性已经提高了处理时间,但是,前面说过,邮件和短信对我正常的使用网站没有任何影响,客户端没有必要等着其发送完成才显示注册成功,应该是写入数据库后就返回.

1.3 引入消息队列

引入消息队列后,把发送邮件/短信等不是必须的业务逻辑异步处理.  
![Image 1][]![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 1][]

由此可以看出,引入消息队列后,用户的响应时间就等于写入数据库的时间+写入消息队列的时间(可以忽略不计),引入消息队列后处理后,响应时间是串行的3倍,是并行的2倍.

#### 2. 应用解耦 ####

场景: 双11是购物狂节,用户下单后,订单系统需要通知库存系统,传统的做法就是订单系统调用库存系统的接口.  
![Image 1][]![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 2][]

该做法的缺点

*  当库存系统出现故障时,订单就会失败;
 *  订单系统和库存系统高耦合.

引入消息队列

![Image 1][]![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 3][]

*  订单系统:用户下单后,订单系统完成持久化处理,将消息写入消息队列,返回用户订单下单成功.
 *  库存系统:订阅下单的消息,获取下单消息,进行库操作.
 *  就算库存系统出现故障,消息队列也能保证消息的可靠投递,不会导致消息丢失.

#### 3. 流量削峰 ####

流量削峰一般在秒杀活动中应用广泛.

场景:秒杀活动,一般会因为流量过大,导致应用挂掉,为了解决这个问题,一般在应用前端加入消息队列.

作用:

1️⃣.可以控制活动人数,超过一定阀值的订单直接丢弃;

2️⃣.可以缓解短时间的高流量压垮应用(应用程序按自己的最大处理能力获取订单).

![Image 1][]![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 4][]

1️⃣.用户的请求:服务器收到之后,首先写入消息队列,假如消息队列长度超过最大值,则直接抛弃用户请求或跳转到错误页面;

2️⃣.秒杀业务根据消息队列中的请求信息,再做后续处理.

### 三. RabbitMQ的系统架构 ###

![Image 1][]![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 5][]

#### 1. RabbitMQ核心概念 ####

1️⃣.Broker:它提供一种传输服务,它的角色就是维护一条从生产者到消费者的路线,保证数据能按照指定的方式进行传输.

2️⃣.Exchange: 消息交换机,它指定消息按什么规则,路由到哪个队列.

3️⃣.Queue:消息的载体,每个消息都会被投到一个或多个队列.

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

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

6️⃣.vhost:虚拟主机,一个broker里可以有多个vhost,用作不同用户的权限分离.

7️⃣.Producer:消息生产者,就是投递消息的程序.

8️⃣.Consumer:消息消费者,就是接受消息的程序.

9️⃣.Channel:消息通道,在客户端的每个连接里,可建立多个channel.

### 四. RabbitMQ 消息发送原理 ###

首先我们必须连接到RabbitMQ才能发布和消费消息,那怎么连接和发送消息的呢?

首先我们的应用程序和Rabbit Server之间要创建一个TCP连接.一旦TCP连接打开,并通过了认证,认证就是你试图连接RabbitMQ之前发送的RabbitMQ服务器连接信息和用户名和密码,有点像程序连接数据库,一旦认证通过你的应用程序和Rabbit就创建了一条AMQP信道(Channel).

信道是创建在“真实”的TCP连接上的虚拟连接,AMQP命令都是通过信道发送出去的.每个信道都有一个唯一的ID,不论是发布消息,订阅队列或者消费消息都是通过信道完成的.

#### 为什么不通过TCP直接发送命令？ ####

对于操作系统来说创建和销毁TCP会话是非常昂贵的开销,假设高峰期每秒有成千上万条连接,每个连接都要创建一条TCP会话,这就造成了TCP连接的巨大浪费,而且操作系统每秒能创建的TCP也是有限的,因此很快就会遇到系统瓶颈.

如果我们每个请求都使用一条TCP连接,既满足了性能的需要,又能确保每个连接的私密性,这就是引入信道概念的原因.  
![Image 1][]![20190612231721380.png][]

### 五. 消息持久化 ###

#### 1. 消息持久化概述 ####

RabbitMQ的消息队列和交换器,在默认情况下重启服务器之后会导致消息丢失,我们可以通过消息持久化来保证RabbitMQ在重启之后消息不丢失.

当你把消息发送到RabbitMQ服务器的时候,你需要选择是否进行持久化,但这并不能保证RabbitMQ能从崩溃中恢复,想要RabbitMQ消息能恢复必须满足3个条件:

*  投递消息的时候durable设置为true,消息持久化;
 *  消息已经到达持久化交换器上;
 *  消息已经到达持久化的队列;

#### 2. 持久化工作原理 ####

RabbitMQ会将你的持久化消息写入到磁盘上的持久化日志文件中,等消息被消费之后,RabbitMQ会把这条消息标识为等待垃圾回收的消息.

#### 3. 持久化的缺点 ####

消息持久化的优点显而易见,但缺点也很明显,那就是性能.因为要写入硬盘中,所以性能要比写入内存中低很多,从而降低了服务器的吞吐量,尽管使用SSD硬盘可以使事情得到缓解,但它仍然降低了RabbitMQ的性能,当消息成千上万条要写入磁盘的时候,性能是很低的.

所以使用者要根据自己的情况,选择适合自己的方式.

### 六. 任务分发机制 ###

#### 1. Round-robin dispathching 循环分发 ####

RabbbitMQ的分发机制非常适合扩展,而且它是专门为并发程序设计的,如果现在load加重,那么只需要创建更多的Consumer来进行任务处理.

#### 2. Message acknowledgment 消息确认 ####

为了保证数据不被丢失,RabbitMQ支持消息确认机制,为了保证数据能被正确处理而不仅仅是被Consumer收到,那么我们不能采用no-ack,而应该是在处理完数据之后发送ack.

在处理完数据之后发送ack,就是告诉RabbitMQ数据已经被接收,处理完成,RabbitMQ可以安全的删除它了.

如果Consumer退出了但是没有发送ack,那么RabbitMQ就会把这个Message发送到下一个Consumer，这样就保证在Consumer异常退出情况下数据也不会丢失.

RabbitMQ没有用到超时机制.RabbitMQ仅仅通过Consumer的连接中断来确认该Message并没有正确处理,也就是说RabbitMQ给了Consumer足够长的时间做数据处理.

如果忘记ack,那么当Consumer退出时,Mesage会重新分发,然后RabbitMQ会占用越来越多的内存.

### 七. Message durability消息持久化 ###

要持久化队列queue的持久化需要在声明时指定durable=True.

这里要注意,队列的名字一定要是Broker中不存在的,否则不能改变此队列的任何属性.

队列和交换机有一个创建时候指定的标志durable,durable的唯一含义就是具有这个标志的队列和交换机会在重启之后重新建立,它不表示说在队列中的消息会在重启后恢复.

#### 消息持久化的组成 ####

1️⃣. exchange持久化,在声明时指定durable = true

hannel.ExchangeDeclare(ExchangeName, "direct", durable: true, autoDelete: false, arguments: null);//声明消息队列,且为可持久化的

2️⃣.queue持久化,在声明时指定durable = true

channel.QueueDeclare(QueueName, durable: true, exclusive: false, autoDelete: false, arguments: null);//声明消息队列,且为可持久化的

3️⃣.消息持久化,在投递时指定delivery\_mode = 2(1是非持久化).

channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());

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

注意：

一旦创建了队列和交换机,就不能修改其标志了,例如,创建了一个non-durable的队列,然后想把它改变成durable的,唯一的办法就是删除这个队列然后重现创建.

### 八. Fair dispath 公平分发 ###

你可能也注意到了,分发机制不是那么优雅,默认状态下,RabbitMQ将第n个Message分发给第n个Consumer.n是取余后的,它不管Consumer是否还有unacked Message,只是按照这个默认的机制进行分发.

那么如果有个Consumer工作比较重,那么就会导致有的Consumer基本没事可做,有的Consumer却毫无休息的机会,那么,Rabbit是如何处理这种问题呢?

![Image 1][]![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 6][]

通过basic.qos方法设置prefetch\_count=1,这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message.换句话说,在接收到该Consumer的ack前,它不会将新的Message分发给它.

channel.basic_qos(prefetch_count=1)

注意:

这种方法可能会导致queue填满.当然,这种情况下你可能需要添加更多的Consumer,或者创建更多的virtualHost来细化你的设计.

### 九. 分发到多个Consumer ###

#### 1. Exchange ####

先来温习以下交换机路由的几种类型:

Direct Exchange:

直接匹配,通过Exchange名称+RountingKey来发送与接收消息.

Fanout Exchange:

广播订阅,向所有的消费者发布消息,但是只有消费者将队列绑定到该路由器才能收到消息,忽略Routing Key.

Topic Exchange：

主题匹配订阅,这里的主题指的是RoutingKey.RoutingKey可以采用通配符,如:\*或\#，RoutingKey命名采用.来分隔多个词,只有消息这将队列绑定到该路由器且指定RoutingKey符合匹配规则时才能收到消息;

Headers Exchange:

消息头订阅,消息发布前,为消息定义一个或多个键值对的消息头,然后消费者接收消息同时需要定义类似的键值对请求头:(如:x-mactch=all或者x\_match=any)，只有请求头与消息头匹配,才能接收消息,忽略RoutingKey.

默认的exchange:如果用空字符串去声明一个exchange，那么系统就会使用”amq.direct”这个exchange.我们创建一个queue时,默认的都会有一个和新建queue同名的routingKey绑定到这个默认的exchange上去.

channel.BasicPublish("", "TaskQueue", properties, bytes);

因为在第一个参数选择了默认的exchange，而我们申明的队列叫TaskQueue，所以默认的，它新建了一个也叫TaskQueue的routingKey，并绑定在默认的exchange上，导致了我们可以在第二个参数routingKey中写TaskQueue，这样它就会找到定义的同名的queue，并把消息放进去。

如果有两个接收程序都是用了同一个queue和相同的routingKey去绑定direct exchange的话，分发的行为是负载均衡的，也就是说第一个是程序1收到，第二个是程序2收到，以此类推。

如果有两个接收程序用了各自的queue,但使用相同的routingKey去绑定direct exchange的话,分发的行为是复制的,也就是说每个程序都会收到这个消息的副本,行为相当于fanout类型的exchange.

#### 2. Bindings 绑定 ####

绑定其实就是关联了exchange和queue,或者这么说:queue对exchange的内容感兴趣,exchange要把它的Message deliver到queue.

#### 3. Direct exchange ####

Driect exchange的路由算法非常简单:通过bindingkey的完全匹配,可以用下图来说明.  
![Image 1][]![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 7][]

Exchange和两个队列绑定在一起,Q1的bindingkey是orange,Q2的binding key是black和green.

当Producer publish key是orange时,exchange会把它放到Q1上,如果是black或green就会到Q2上,其余的Message被丢弃.

#### 4. Multiple bindings ####

多个queue绑定同一个key也是可以的,对于下图的例子,Q1和Q2都绑定了black,对于routing key是black的Message,会被deliver到Q1和Q2，其余的Message都会被丢弃.  
![Image 1][]![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 8][]

#### 5. Topic exchange ####

Message的routing\_key是有限制的,不能使任意的.格式是以点号“.”分割的字符表,比如：”stock.usd.nyse”, “nyse.vmw”, “quick.orange.rabbit”.你可以放任意的key在routing\_key中,当然最长不能超过255 bytes.

对于routing\_key,有两个特殊字符:

\*(星号)代表任意一个单词;

\#(hash)代表0个或多个单词.

![Image 1][]![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 9][]

Producer发送消息时需要设置routing\_key,routing\_key包含三个单词和连个点号o,第一个key描述了celerity(灵巧),第二个是color(色彩),第三个是物种:  
在这里我们创建了两个绑定: Q1 的binding key 是".orange.";  
Q2 是"..rabbit" 和 "lazy.\#";

Q1感兴趣所有orange颜色的动物;  
Q2感兴趣所有rabbits和所有的lazy的.

例子:rounting\_key 为 “quick.orange.rabbit”将会发送到Q1和Q2中  
rounting\_key 为”lazy.orange.rabbit.hujj.ddd”会被投递到Q2中,\#匹配0个或多个单词.

### 十. 消息序列化 ###

RabbitMQ使用ProtoBuf序列化消息,它可作为RabbitMQ的Message的数据格式进行传输.  
由于是结构化的数据,这样就极大的方便了Consumer的数据高效处理,当然也可以使用XML,与XML相比,ProtoBuf有以下优势:

1️⃣.简单;

2️⃣.size小了3-10倍;

3️⃣.速度快了20-100倍;

4️⃣.易于编程;

5️⃣.减少了语义的歧义,ProtoBuf具有速度和空间的优势,使得它现在应用非常广泛.

[Image 1]: 
[20190612231605129.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/20/1768d34076fc4f519c612bde178934d6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/20/4f1eb077def64feeaccbd20210c4a9aa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/20/26ccf78b43024a3aaaa72d24db38611d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/20/88617495554c48f6b61a883ff3efdf89.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/20/a64c3054d4184c67919ab401c06c406c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/20/2f13eb75efd44aeaa051714291c107b5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 5]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/20/da2f35e46d6f41039b6c6b8df43a4245.png
[20190612231721380.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/20/cea12d9227164f64bcdcf877cff1562e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 6]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/20/bd406815140341c791a87c194feb1ab9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 7]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/20/07eec757576a4b399652ef84638d7ae3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 8]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/20/3ad759b98765454b8a8108f9ae2b1834.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N5YzAwMDY2Ng_size_16_color_FFFFFF_t_70 9]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/20/e76f13ffd3a646f59afd5392308a2b70.png