RabbitMQ的应用总结

你的名字 2024-04-17 22:05 43阅读 0赞

**RabbitMQ**是一个由erlang开发的AMQP（Advanced Message Queue ）的开源实现。AMQP 的出现其实也是应了广大人民群众的需求，虽然在同步消息通讯的世界里有很多公开标准（如 COBAR的 IIOP ，或者是 SOAP 等），但是在异步消息处理中却不是这样，只有大企业有一些商业实现（如微软的 MSMQ ，IBM 的 Websphere MQ 等），因此，在 2006 年的 6 月，Cisco 、Redhat、iMatix 等联合制定了 AMQP 的公开标准。  
RabbitMQ是由RabbitMQ Technologies Ltd开发并且提供商业支持的。该公司在2010年4月被SpringSource（VMWare的一个部门）收购。在2013年5月被并入Pivotal。其实VMWare，Pivotal和EMC本质上是一家的。不同的是VMWare是独立上市子公司，而Pivotal是整合了EMC的某些资源，现在并没有上市。  
RabbitMQ的官网是http://www.rabbitmq.com  
百度百科amqp协议介绍https://baike.baidu.com/item/AMQP/8354716?fr=aladdin  
**注意:RabbitMQ是采用erlang语言开发的，所以必须有Erlang环境才可以运行**

**Erlang  (高并发应用) **

Erlang编程语言最初目的是进行大型电信交换设备的软件开发，是一种适用于大规模并行处理环境的高可靠性编程语言。随着多核处理器技术的日渐普及，以及互联网、云计算等技术的发展，该语言的应用范围也有逐渐扩大之势。

百度百科介绍:https://baike.baidu.com/item/Erlang%E8%AF%AD%E8%A8%80/20864044?fr=aladdin

比较图示：  初衷理念实现抗高并发语言

![1179709-20190901121837628-1528602051.png][]

不同的项目 不同的 路径，独立的virtualhost，相互进行隔离：  客户端连接时候需要指定virtual host地址 。

更加解耦 相互进行隔离  类似于每个项目都有不同的数据库一样。

![1179709-20190901122105459-1295343873.png][]

添加virtual host

![1179709-20181110135813596-1946748860.png][]

指定某个用户的 Virtual Host

![1179709-20181110140054695-517432697.png][]

![1179709-20181110140154438-1686646626.png][]

**AMQP**（高级消息队列协议）是一个异步消息传递所使用应用层协议规范，为面向消息中间件设计，基于此协议的客户端与消息中间件可以无视消息来源传递消息，不受客户端、消息中间件、不同的开发语言环境等条件的限制；

**涉及概念解释：**   
**Server(Broker):**接收客户端连接，实现AMQP协议的消息队列和路由功能的进程；  
**Virtual Host：**虚拟主机的概念，类似权限控制组，一个Virtual Host里可以有多个Exchange和Queue。     
**Exchange:**交换机，接收生产者发送的消息，并根据Routing Key将消息路由到服务器中的队列Queue。  
**ExchangeType:**交换机类型决定了路由消息行为，RabbitMQ中有三种类型Exchange，分别是fanout、direct、topic；  
**Message Queue：**消息队列，用于存储还未被消费者消费的消息；  
**Message：**由Header和body组成，Header是由生产者添加的各种属性的集合，包括Message是否被持久化、优先级是多少、由哪个Message Queue接收等；body是真正需要发送的数据内容；  
**BindingKey：**绑定关键字，将一个特定的Exchange和一个特定的Queue绑定起来。

## RabbitMQ几种工作模式: ##

**RabbitMQ点对点模式:**

1、点对点模式  一对一模式。  一个生产者投递消息给队列 只能允许有一个消费者进行消费    如果集群的话 会进行均摊消费   服务器配置不一样 均摊就不优了

![1179709-20190901123433910-1743920122.png][]

长连接 不用三次握手之类的 提高传输效率     但是长连接占服务器带宽

推：  消费者已经启动了，建立长连接，一旦生产者向队列投递消息会立马推送给消费者

取： 生产者先投递消息队列进行缓存，这时候消费者再次启动时候 ，就会向队列获取消息。

点对点模式代码: [RabbitMQ整合Spring Booot点对点模式][RabbitMQ_Spring Booot]

### **RabbitMQ手动模式和**自动应答模式 ###

1.了确保消息不会丢失，RabbitMQ支持消息应答。消费者发送一个消息应答，告诉RabbitMQ这个消息已经接收并且处理完毕了。RabbitMQ就可以删除它了。  
   2. 如果一个消费者挂掉却没有发送应答，RabbitMQ会理解为这个消息没有处理完全，然后交给另一个消费者去重新处理。这样，你就可以确认即使消费者偶尔挂掉也不会丢失任何消息了。  
   3. 没有任何消息超时限制；只有当消费者挂掉时，RabbitMQ才会重新投递。即使处理一条消息会花费很长的时间。  
   4. 消息应答是默认打开的。我们通过显示的设置autoAsk=true关闭这种机制。现即自动应答开，一旦我们完成任务，消费者会自动发送应答。通知RabbitMQ消息已被处理，可以从内存删除。如果消费者因宕机或链接失败等原因没有发送ACK（不同于ActiveMQ，在RabbitMQ里，消息没有过期的概 念),则RabbitMQ会将消息重新发送给其他监听在队列的下一个消费者。

**应答模式:**

自动应答： 不在乎消费者对消息处理是否成功，都会告诉队列删除消息。如果处理消息失败，实现自动补偿（队列投递过去 重新处理）。

手动应答： 消费者处理完业务逻辑，手动返回ack（通知）告诉队列处理完了，队列进而删除消息。

**实现思路:**

*  生产者端代码不变，消费者端代码这部分就是用于开启手动应答模式的。

channel.basicConsume(QUEUE\_NAME, false, defaultConsumer);  
         注：第二个参数值为false代表关闭RabbitMQ的自动应答机制，改为手动应答。

*  在处理完消息时，返回应答状态，true表示为自动应答模式。

channel.basicAck(envelope.getDeliveryTag(), false);

自动应答： 不在乎消费者对消息处理是否成功，都会告诉队列删除消息。如果处理消息失败，实现自动补偿（队列投递过去 重新处理）。

手动应答： 消费者处理完业务逻辑，手动返回ack（通知）告诉队列处理完了，队列进而删除消息。

[RabbitMQ整合Spring Booot【消费者应答模式】][RabbitMQ_Spring Booot 1]

## RabbitMQ队列形式 ##

###   公平队列 ###

*       目前消息转发机制是平均分配，这样就会出现俩个消费者，奇数的任务很耗时，偶数的任何工作量很小，造成的原因就是近当消息到达队列进行转发消息。并不在乎有多少任务消费者并未传递一个应答给RabbitMQ。仅仅盲目转发所有的奇数给一个消费者，偶数给另一个消费者。
 *       为了解决这样的问题，我们可以使用basicQos方法，传递参数为prefetchCount= 1。这样告诉RabbitMQ不要在同一时间给一个消费者超过一条消息。

** 总结：**

只有在 消费者空闲的时候会发送下一条信息。调度分发消息的方式，也就是告诉RabbitMQ每次只给消费者处理一条消息，也就是等待消费者处理完毕并自己对刚刚处理的消息进行确认之后，才发送下一条消息，防止消费者太过于忙碌，也防止它太过去清闲。

** 公平队列原理：**队列服务器向消费者发送消息的时候，消费者采用手动应答模式，队列服务器必须要收到消费者发送ack结果通知，才会发送下一个消息。（快的处理的多，消费的多）

**使用背景:**

*  服务器能力不同，能者多劳。 均摊模式的话，都处理相同数量的
 *   消息队列 发出去的消息被消费完了 然后收到 ack包 才可以继续发给他

![1179709-20190901145702173-129218201.png][]

通过 设置channel.basicQos(1); 开发:   [RabbitMQ整合Spring Booot【公平队列】][RabbitMQ_Spring Booot 2]

## RabbitMQ死信队列 ##

**关于RabbitMQ死信队列**

*  死信队列 听上去像 消息“死”了  。其实也有点这个意思，死信队列 是当消息在一个队列，因为下列原因：

1.  消息被拒绝（basic.reject/ basic.nack）并且不再重新投递 requeue=false
2.  消息超期 (rabbitmq  Time-To-Live -> messageProperties.setExpiration())，消息TTL过期。
3.  队列超载，队列达到最大长度（队列满了，无法再添加数据到mq中）

变成了 “死信” 后 ，被重新投递（publish）到另一个Exchange ，该Exchange 就是DLX 。

然后该Exchange 根据绑定规则，转发到对应的 队列上， 监听该队列 ，就可以重新消费。

说白了 就是 没有被消费的消息 ，换个地方重新被消费

生产者   -->  消息 --> 交换机  --> 队列  --> 变成死信  --> DLX交换机 -->队列 --> 消费者

**应用场景分析：**

在定义业务队列的时候，可以考虑指定一个死信交换机，并绑定一个死信队列，当消息变成死信时，该消息就会被发送到该死信队列上，这样就方便我们查看消息失败的原因了

死信队列 听上去像 消息“死”了 ，其实也有点这个意思，  
死信队列 是 当消息在一个队列 因为下列原因:  
1.消息被拒绝（basic.reject或basic.nack）并且requeue=false.  
2.消息TTL过期  
3.队列达到最大长度（队列满了，无法再添加数据到mq中）

**应用场景分析：**

*   在定义业务队列的时候，可以考虑指定一个死信交换机，并绑定一个死信队列，当消息变成死信时，该消息就会被发送到该死信队列上，这样就方便我们查看消息失败的原因了

channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false); 丢弃消息

*  如果高并发情况到来  某一个队列比如邮件队列满了 或者异常  或者消息过期 或者消费者拒绝消息

**如何使用死信交换机呢？**

定义业务（普通）队列的时候指定参数

1.  x-dead-letter-exchange: 用来设置死信后发送的交换机
2.  x-dead-letter-routing-key：用来设置死信的routingKey

![1179709-20190901172810616-374006860.png][]

1.   邮件队列 绑定一个死信交换机 ，一旦邮件队列满了的情况下 ，为了防止数据丢失情况  ，消息不再邮件队列存放了，放到死信交换机。
2.  然后交给私信邮件队列。
3.  最终交给死信消费者。

步骤：

1.    创建死信交换机 、死信队列、 并且绑定
2.    之前的队列没有绑定死信队列和死信交换机 不能做更改绑定死信交互机
3.    之前创建好的邮件队列 删除掉  已经创建好的队列不能做更改  交换机也清理掉

[RabbitMQ整合Spring Booot【死信队列】][RabbitMQ_Spring Booot 3]

## Exchange ##

**这个可能是消息队列中最重要的队列了，其他的都是在它的基础上进行了扩展。**

**关于交换机：**

1.  生产者发送消息不会向传统方式直接将消息投递到队列中，而是先将消息投递到交换机中，在由交换机转发到具体的队列，队列在将消息以推送或者拉取方式给消费者进行消费，这和我们之前学习Nginx有点类似。交换机的作用根据具体的路由策略分发到不同的队列中。

**交换机有四种类型:**

1.  ** Direct exchange（直连交换机）:**是根据消息携带的路由键（routing key）将消息投递给对应队列的。
2.  ** Fanout exchange（扇型交换机）:**将消息路由给绑定到它身上的**所有队列。**
3.  ** Topic exchange（主题交换机）:**队列通过路由键绑定到交换机上，然后，交换机根据消息里的路由值，将消息路由给一个或多个绑定队列。
4.  ** Headers exchange（头交换机）:**类似主题交换机，但是头交换机使用多个消息属性来代替路由键建立路由规则。通过判断消息头的值能否与指定的绑定相匹配来确立路由规则。

**过程:**

一个生产者发送消息  ----> 到交换机  ---->  到队列(每个队列绑定到交换机上)  ----> 到消费者(每个消费者有自己的队列)

**功能实现：**一个生产者发送消息，多个消费者获取消息（同样的消息）, 包括：

*     一个生产者
 *    一个交换机
 *    多个队列
 *    多个消费者

![1179709-20181110225834666-1774156298.png][]

### **1. Direct****Exchange ：** ###

### **   （1） 路由key模式:** ###

**场景**：

*  生产者发送消息到交换机并指定一个路由key，
 *  消费者队列绑定到交换机时要指定路由key（key匹配就能接受消息，key不匹配就不能接受消息）

**例如**：

我们可以把路由key设置为insert ，那么消费者队列key指定包含insert才可以接收消息，消费者队列key定义为update或者delete就不能接收消息。很好的控制了更新，插入和删除的操作。

**注：**

前面做的demo中RoutingKey设置的空。

![1179709-20181111204917630-725984167.png][]

**代码实现:**

RoutingKey有值的时候，那么 经过消息队列之后，需要在经过RoutingKey进行判断决定消费者。

[RabbitMQ整合Spring Booot【Exchange-路由key模式】][RabbitMQ_Spring Booot_Exchange-_key]

### **（1） Topics模式:** ###

*     此模式实在路由key模式的基础上，使用了通配符来管理消费者接收消息。
 *     生产者P发送消息到交换X，type=topic。交换机根据绑定队列的routing key的值进行通配符匹配。

-    符号\#：匹配一个或者多个词lazy.\# 可以匹配lazy.irs或者lazy.irs.cor。

-    符号\*：只能匹配一个词lazy.\* 可以匹配lazy.irs或者lazy.cor。

![1179709-20181111225201052-1079657782.png][]

关于 通配符 “，” 和 “\#” 的使用: [RabbitMQ整合Spring Booot【Exchange-Topics模式】][RabbitMQ_Spring Booot_Exchange-Topics]

** (3)****Fanout Exchange:**

RabbitMQ发布与订阅原理 Exchange Fanout模式：

案例： 用户注册 ---> 发送邮件 --->发送短信

**![1179709-20190901174910904-1153054331.png][]**

[RabbitMQ整合Spring Booot【Exchange-Fanout模式】][RabbitMQ_Spring Booot_Exchange-Fanout]

## RabbitMQ生产者事务确认机制  ##

**背景:**

生产者发送消息出去之后，不知道到底有没有发送到RabbitMQ服务器， 默认是不知道的。而且有的时候我们在发送消息之后，后面的逻辑出问题了，我们不想要发送之前的消息了，需要撤回该怎么做?

**解决方案:**

**AMQP 事务机制**

事务模式:

*   txSelect 将当前channel设置为transaction模式 （开启事务）
 *   txCommit 提交当前事务   （提交事务）
 *  txRollback 事务回滚   （回滚事务）

注: RabbitMQ支持消息持久化机制，把消息持久化到硬盘上。如果RabbitMQ服务器宕机了，消息不会丢失。

[RabbitMQ整合Spring Booot【生产者事务确认机制】][RabbitMQ_Spring Booot 4]

** 方案二：Confirm 模式**

队列和消费者建立长连接，推送或者拉取形式。

消费者通过自动应答或者手动应答，队列服务器等待应答结果，如果没有应答结果那么保留给下一个消费者。

![1179709-20190901171000506-214026074.png][]

## RabbitMQ消费者重试机制  ##

消费者运行报错时候，会进行重试。

[RabbitMQ整合Spring Booot【消费者补偿幂等问题】][RabbitMQ_Spring Booot 5]

## RabbitMQ解决分布式事务问题: ##

[RabbitMQ解决分布式事务][RabbitMQ]

# #

## 小结： ##

1.点对点(简单)的队列  
2.工作(公平性)队列模式  
3.发布订阅模式  
4.路由模式Routing（Direct）  
5.通配符模式Topics

[1179709-20190901121837628-1528602051.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/a9a2d26297214196a5d5a9b28afc5184.png
[1179709-20190901122105459-1295343873.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/52b632a1f54c448db9b25ab0851981b5.png
[1179709-20181110135813596-1946748860.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/b79df074b4b64c5da0bf004a2d8c5e67.png
[1179709-20181110140054695-517432697.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/5a97cff044884a8088bec1aaa84c79a4.png
[1179709-20181110140154438-1686646626.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/a0886764b4d84ffa89302fc9e863ff9f.png
[1179709-20190901123433910-1743920122.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/dc075b24b5dc477780d27949f146e3b7.png
[RabbitMQ_Spring Booot]: https://www.cnblogs.com/toov5/p/11441982.html
[RabbitMQ_Spring Booot 1]: https://www.cnblogs.com/toov5/p/11442296.html
[1179709-20190901145702173-129218201.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/8e999a2d38d649f49d33c241d811167a.png
[RabbitMQ_Spring Booot 2]: https://www.cnblogs.com/toov5/p/9940566.html
[1179709-20190901172810616-374006860.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/84d736b4ccd249d8ab38321b101377c8.png
[RabbitMQ_Spring Booot 3]: https://www.cnblogs.com/toov5/p/10288260.html
[1179709-20181110225834666-1774156298.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/6746ff04767b4f63b3dbd51b49cdc0e1.png
[1179709-20181111204917630-725984167.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/bb950a6a52cb43a9a58f6b1d2bbc1f6d.png
[RabbitMQ_Spring Booot_Exchange-_key]: https://www.cnblogs.com/toov5/p/9944019.html
[1179709-20181111225201052-1079657782.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/eb8298f1c271455e83c75b3c20c35f38.png
[RabbitMQ_Spring Booot_Exchange-Topics]: https://www.cnblogs.com/toov5/p/9944291.html
[1179709-20190901174910904-1153054331.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/29e3db493e384e609e24de8e1223d2c0.png
[RabbitMQ_Spring Booot_Exchange-Fanout]: https://www.cnblogs.com/toov5/p/11029472.html
[RabbitMQ_Spring Booot 4]: https://www.cnblogs.com/toov5/p/9945017.html
[1179709-20190901171000506-214026074.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/ba2c2cc568df49a48a1cb0cff54a91bf.png
[RabbitMQ_Spring Booot 5]: https://www.cnblogs.com/toov5/p/10287183.html
[RabbitMQ]: https://www.cnblogs.com/toov5/p/10289999.html