精通RabbitMQ之AMQP杂谈

川长思鸟来 2022-04-03 14:50 220阅读 0赞

### 精通RabbitMQ之AMQP杂谈 ###

*  何为AMQP
 *  AMQP能够做什么
 *   *  应用解耦（异步）
     *  异步处理
     *  同步调用
     *  顺序调度
     *  通知分发
     *  高并发缓冲
     *  并发限流
     *  延时任务调度
     *  失败重试

# 何为AMQP #

**AMQP，即Advanced Message Queuing Protocol（高级消息队列协议），它的定义使得符合要求的客户端应用（application）和消息中间件代理（messaging middleware broker）之间可以进行通信。它是应用层协议的一个开放标准，是面向消息的中间件设计，用于在分布式系统中存储转发消息。**

要想了解清楚为什么会有AMQP以及它能够为我们做什么，了解它的技术产生和发展历程是非常有必要的，因为**技术的发展从来都是需求的推动。**

Message Queue的需求由来已久，最早可以追溯到80年代的金融交易。要想了解Message Queue的需求，我们有必要扯一下系统架构演化历程，一开始网站业务没有太多人访问，只需要一台服务器就绰绰有余，后面为了应对日益复杂的业务场景，任何强大的单一服务器都满足不了大型网站持续增长的业务需求。必须通过使用分而治之的手段将整个网站业务分成不同的业务，随着发展，业务拆分越来越小，数据存储系统越来越庞大（例如数据库经过读写分离后，从一台服务器拆分成两台服务器，再到后面演化为多主多从），但是随着网站业务的发展依然不能满足需求，这时需要使用分布式数据库。文件系统也一样，需要使用分布式文件系统，也就成了我们现在的**分布式环境**。但是此时也演化出了一个新问题分布式环境下的通信问题。于是**AMQP就被设计出来用于解决分布式环境下的通信问题**。

# AMQP能够做什么 #

## 应用解耦（异步） ##

随着我们业务开发的不断拓展，业务必然会越来越复杂，从而不得不对我们的业务进行拆分，但最终形成分布式系统必然也会出现系统间通信的问题，例如下面的上下游相互调用的问题。那么这个问题在传统的场景下能不能解决呢？

其实也可以，传统场景下，我们可以通过上下游系统间的接口调用来解决这个问题。  
![上下游接口调用][20181223101520946.png]

但是，能解决不代表完美的解决了。上下游系统接口调用必然带来上下游系统严重耦合的问题，这里面就至少存在2方面的问题：

1.  假如下游系统出现问题，上游系统无法调用下游系统，上游系统的处理自然也无法完成，整个系统就会崩溃。这在实际开发中是一个巨大的灾难。举个最简单的例子，订单系统完成订单预定之后需要通知库存系统去发货。此时库存系统由于网络故障无法访问，订单系统由于无法通知到库存系统只能拒绝所有的订单。
2.  假如下游系统遇到架构调整问题，上游系统也必须被动同步跟着调整。（比如，下游系统的IP调整了，上游系统的接口地址就需要调整）。

问题场景图如下：  
![上下游互相调用故障][20181223101901892.png]

那么上下游系统严重耦合的问题，应该怎么解决呢？

此时，AMQP的**消息中间件**就可以派上用场了。上游系统处理完成之后，会把数据消息写入消息队列中，它的处理就算结束了；下游系统可以订阅消息队列，也可以自由的从消息队列拉取消息，上下游毫不关联。例如，订单系统接收到了订单之后，只用把通知消息写入消息队列，此时就算库存系统故障也丝毫不影响用户下单，待库存系统恢复后消息队列可以再推送消息给库存系统（也可以自己去取）。

逻辑设计图如下：  
![mq中间件解决][mq]

应用解耦可以说是MQ需要解决的最根本的问题，后续发展的过程中MQ能解决的问题越来越多，但都有应用解耦的概念在里面，所以我们这里要重点说明。让大家有这个思想。

## 异步处理 ##

异步处理 ，其实没什么好说的，上面举到的订单系统异步通知库存系统就是最好的例子，但是这里需要注意，并不代表所有的异步任务都需要消息队列，对于精度不高（丢失也无所谓）或者能通过线程池另起线程处理的最好还是通过多线程处理（例如欢迎注册邮件）。但是如果精度要求很高（需要失败恢复重新拉起的），或者超过线程池最高并发处理线程数量的问题，MQ可以很好的解决这个问题，MQ本身就具有**持久化存储消息**的功能。

## 同步调用 ##

何为同步调用? 对比于上面我们说到的订单系统通知库存系统的例子，此时，我们是不需要结果回执的，库存系统可以慢慢处理，但是假如我们需要异步处理系统的处理结果回执呢？比如订单系统是需要支付系统的支付确认回执的。此时简单的丢消息到队列中很显然就不可靠了，当然AMQP也可以很完好的支持同步调用，后面我用一段具体的demo来掩饰，这里就不做过多说明了，当然随着技术的发展，同步调用的实现现在更多的由微服务来实现，使用AMQP来实现同步调用已经退出了历史舞台。

## 顺序调度 ##

何为顺序调度？ 其实就是利用一个队列的先进先出原则去调度有严格先后顺序的任务。例如，先通知仓库打包才能通知物流取货。

## 通知分发 ##

通知分发 的场景也很简单，主要利用`Exchange`的`fanout`／`topic`类型进行发布订阅，例如订单系统的下单消息几乎要分发给所有的其他系统，我们就可以让所有的系统都订阅订单系统的的消息。

## 高并发缓冲 ##

何为高并发缓冲？ 这里主要利用消息队列作为消息存储容器的特性，高并发的情况下可能1秒种之内收到几十万的请求量，这个时候业务系统处理不过来就挂了，参考12306／选课系统等。这个时候，我们可以把所有的请求全部存入到队列（队列的存储能力理论上无限的，主要受制于空间），然后再批量取出能处理数量的消息处理。

## 并发限流 ##

何为并发限流？ 这里主要利用消息队列的设置队列消息的长度(容量)特性，高并发的情况下，只接受指定数量的请求，队列满了就直接拒绝。

## 延时任务调度 ##

何为延时任务调度？ 这里主要利用的是消息队列中的消息过期后进入到死信队列的特性，大致实现思路就是为消息设置等同延时的有效期，消息过期后进入到死信队列，然后再到死信队列取消息处理。

## 失败重试 ##

何为失败重试？ 这里主要利用的是消息队列的持久化特性，假如此时，我们有很多任务需要异步处理，我们可以利用线程池另起线程处理，但是线程处理过程中挂掉就等于处理失败并且不可恢复，这里我们可能无法接受，这个时候消息队列的持久化就可以发挥作用了，我们可以将消息持久化在队列中，只用当消息队列收到了处理系统的处理完成确认回执才会将我们的消息从队列中移除。

[20181223101520946.png]: /images/20220403/604bab41ca174be9acbbc7919a004f0c.png
[20181223101901892.png]: /images/20220403/df9ecf670f914c62b432494bf035351c.png
[mq]: /images/20220403/6b88f9b5eedc44b0bef4a9bdff422fa4.png