RabbitMQ编程基本介绍

迈不过友情╰ 2023-02-26 11:22 2阅读 0赞

# 简介 #

RabbitMQ是开源的AMQP（高级消息队列协议）协议的标准实现，服务器端用Erlang语言编写，支持多种客户端，用于在分布式系统中存储和转发消息，在易用性、扩展性（多种交换机可供选择）、高可用性方面表现不俗。

**消息中间件作用**

1.  冗余〈存储)：有些情况下处理数据的过程会失败，造成数据丢失，可使用消息中间件进行数据持久化；
2.  扩展性:消息中间件解耦了应用的处理过程，所以提高消息入队和处理的效率是很容易的，只要另外增加处理过程即可，不需要改变代码，也不需要调节参数。
3.  削峰: 在访问量剧增的情况下，程序不会因为突发的超负荷请求而崩溃。
4.  可恢复性: 当系统一部分组件失效时，不会影响到整个系，消息中间件降低了进程间的耦合度，所以即使 个处理消息的进程挂掉，加入消息中间件中的消息仍然可以在系统恢复后进行处理。
5.  顺序保证: 在大多数使用场景下，数据处理的顺序很重要，大部分消息中间件支持 定程度上的顺序性。
6.  缓冲: 在任何重要的系统中，都会存在需要不同处理时间的元素。消息中间件通过 个缓冲层来帮助任务最高效率地执行，写入消息中间件的处理会尽可能快速 该缓冲层有助于控制和优化数据流经过系统的速度。
7.  异步通信: 在很多时候应用不想也不需要立即处理消息 消息中间件提供了异步处理机制，允许应用把 些消息放入消息中间件中，但并不立即处理它，在之后需要的时候再慢慢处理

本文将着重讲解rmq的基本概念及编程指引（java）

# 一、对比ActiveMQ（协议、使用场景） #

<table> 
 <thead> 
  <tr> 
   <th align="left">ActiveMq</th> 
   <th align="left">RabbitMQ</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">ActiveMq，传统的消息队列，使用Java语言编写。基于JMS（Java Message Service），采用多线程并发，资源消耗比较大。支持P2P和发布订阅两种模式。</td> 
   <td align="left">RabbitMQ，基于AMQP协议实现，支持多种场景，社区活跃量大。高性能，高可用，支持海量数据。</td> 
  </tr> 
  <tr> 
   <td align="left">JMS提供了两种消息模型，peer-2-peer(点对点)以及publish-subscribe（发布订阅）模型。当采用点对点模型时，消息将发送到一个队列，该队列的消息只能被一个消费者消费。而采用发布订阅模型时，消息可以被多个消费者消费。在发布订阅模型中，生产者和消费者完全独立，不需要感知对方的存在。</td> 
   <td align="left">在AMQP中，消息路由（messagerouting）和JMS存在一些差别，在AMQP中增加了Exchange和binding的角色。producer将消息发送给Exchange，binding决定Exchange的消息应该发送到那个queue，而consumer直接从queue中消费消息。queue和exchange的bind有consumer来决定。--------只有一个消费者</td> 
  </tr> 
 </tbody> 
</table>

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center]

当生产者A想要将消息类型B发送给B1，消息类型C发送给C1时，适合rmq~~~

## 1.1 ActiveMQ ##

是一个完全支持JMS1.1和J2EE1.4规范的JMS Provider实现。  
JMS（Java Message Service）是一个在java标准化组织（JCP）内开发的标准，是开发消息服务的标准和规范，允许应用程序组件基于J2EE平台创建、发送、接收和读取消息。

## 1.2 RabbitMQ ##

是开源的AMQP协议的标准实现，支持多种客户端，如Python、Ruby、.NET、Java、JMS、C、PHP等，支持AJAX。

## 1.3 AMQP vs MQTT协议 ##

### 1.3.1 AMQP ###

AMQP ：Advanced Message Queueing Protocol ，高级消息队列协议。它是**应用层协议 ----面试的时候可以向tcp联系**，为面向消息的中间件设计，基于此协议的客户端与消息中间件可传递消息，并不受产品、开发语言等条件的限制。

AMQP是由各种数据帧组成的协议，包括方法帧、内容帧，心跳帧等，每一帧数据所有的帧都由一个头(header,7个字节),任意大小的负载(payload),和一个检测错误的帧结束（frame-end）字节组成

### 1.3.2 MQTT协议 ###

RMQ除了原生支持AMQP协议外，还支持STOMP、MQTT等多种消息中间件协议

MQTT协议是它是专门为**小设备**设计的。`计算性能不高的设备不能适应AMQP上的复杂操作`，它们需要一种简单而且可互用的方式进行通信。这是MQTT的基本要求，而如今，MQTT是物联网（IOT）生态系统中主要成分之一。

# 二、基本概念 #

## 2.1 概念模型 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 1]

1.  Message：消息，包含消息头和消息体
2.  Publisher：消息的生产者，向交换器发送消息
3.  Exchange：接收生产者发送的消息并将这些消息路由到服务器中的队列
4.  Binding：基于routingkey，绑定exchange和queue，建立联系
5.  Queue：消息队列，保存消息直到发送给消费者
6.  Connection：就是一个TCP的连接
7.  Channel：信道，AMQP 命令都是通过信道发出去
8.  Consumer：消息的消费者，从消息队列中取得消息
9.  Broker: 接收和分发消息的应用，实际就是Exchange+Queue
10. Virtual host: 出于多租户和安全因素设计的，把AMQP的基本组件划分到一个虚拟的分组中，类似于网络中的namespace概念。当多个不同的用户使用同一个RabbitMQ server提供的服务时，可以划分出多个vhost，每个用户在自己的vhost创建exchange／queue等。

### 2.1.1 Connection Vs Channel ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 2]  
无论是生产者还是消费者都需要与Broker建立连接，这个连接就是TCP连接，即Connection。一旦TCP连接建立起来，客户端紧接着可以创建一个AMQP信道（Channel），每个信道都会被指派唯一的ID。信道是建立在Connection之上的虚拟连接，RMQ处理的每条AMQP指令都是通过信道完成的。

### 2.1.2 为何要引入Channel？ ###

试想这样一个场景，一个应用程序中有很多个线程需要从 RabbitMQ 中消费消息，或者生产消息，那么必然需要建立很多个 Connection，也就是多个 TCP 连接。

然而对于操作系统而言，建立和销毁 TCP 连接是非常昂贵的开销，如果遇到使用高峰，性能瓶颈也随之显现。

**RabbitMQ 采用类似 NIO（Non-blocking I/O）的做法，选择 TCP 连接复用，不仅可以减少性能开销，同时也便于管理。**

每个线程把持一个信道，所以信道复用了 Connection 的 TCP 连接。同时 RabbitMQ 可以确保每个线程的私密性，就像拥有独立的连接一样。**当每个信道的流量不是很大时，复用单一的 Connection 可以在产生性能瓶颈的情况下有效地节省 TCP 连接资源。**但是信道本身的流量很大时，这时候多个信道复用一个 Connection 就会产生性能瓶颈，进而使整体的流量被限制了。此时就需要开辟多个 Connection，将这些信道均摊到这些 Connection 中，至于这些相关的调优策略需要根据业务自身的实际情况进行调节。

信道在 AMQP 中是一个很重要的概念，大多数操作都是在信道这个层面展开的。

比如 channel.exchangeDeclare、channel.queueDeclare、channel.basicPublish、channel.basicConsume 等方法。

RabbitMQ 相关的 API 与 AMQP 紧密相连，比如 channel.basicPublish 对应 AMQP 的 Basic.Publish 命令。

[在别的博客看到了通俗的解释，很有趣~~~~~~~~~~~~~~~~~~~：][Link 1]

Publisher：消息的发布者；相当于淘宝卖家；  
Consumer：消息的接收者；相当于淘宝买家；  
Connection： 就是一个TCP的连接。Producer和Consumer都是通过TCP连接到RabbitMQ Server的。相当于卖家和快递公司建立合作协议；  
Channels： 虚拟连接。它建立在上述的TCP连接中。数据流动都是在Channel中进行的。相当于卖家的分店与快递公司之间的生意往来；建立和关闭TCP连接耗资源，影响性能，而且TCP的连接数也有限制，限制了系统处理高并发的能力。但是，在TCP连接中建立Channel是没有上述代价的。对于Producer或者Consumer来说，可以并发的使用多个Channel进行Publish或者Receive；  
Exchange：交换机，将消息路由到指定的消费者；相当于快递公司，将接到的快递集散之后发给各个城市的集散中心；  
RootingKey：消息发送给谁的标示符，用来连接Exchange和queue；相当于快递中的地址，让快递公司知道将快递发给哪个集散中心；  
Queue：消息队列，用于缓存消息的队，Consumer和Procuder都可以创建queue，队列的持久化也可以设置；相当于快递的集散中心，用来暂时存放快递；消费者从队列中取消息；相当于买家从集散中心取快递，多个买家可以从同一个集散中心取快递；相当于多个客户端从队列里取消息；一个客户端也可以创建多个通道从队列里取消息，相当于一个家庭的不同成员去取快递； 程序中就是开启多个线程，通过通道从队列中取消息，实现高并发；

## 2.2 消息路由 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 3]  
AMQP 中消息的路由过程和 Java 开发者熟悉的 JMS 存在一些差别，**AMQP 中增加了 Exchange 和 Binding 的角色**。  
生产者把消息发布到 Exchange 上，消息最终到达队列并被消费者接收，  
Binding 决定交换器的消息应该发送到那个队列，更具体地说是基于exchange和queue绑定的routingkey。  
在rabbitmq提供的可视化管理页面rabbitMQ Management中可以很方便地查看exchange、queue以及routingkey等信息。

## 2.3 常用的交换机类型 ##

交换器分为四种，分别是：**direct**、**fanout**、**topic**和 headers

前面三种分别对应路由模式、发布订阅模式和通配符模式，headers 交换器允许匹配 AMQP 消息的 header 而非路由键，除此之外，header 交换器和 direct 交换器完全一致，但是性能却差很多，因此基本上不会用到该交换器，这里也不详细介绍。

### 2.3.1 定向模式 direct ###

![在这里插入图片描述][20201111141235458.png_pic_center]  
使用场景：它是单一传播路由消息的最佳选择！

direct exchange严格根据消息的路由键（routingkey）来传送消息。它是`单一传播路由消息`的最佳选择，routing key将queue与exchange进行绑定，消息根据routing key分配到指定的queue中；一个rooting key可以绑定多个queue，多个rooting key也可以绑定到同一个queue上面。

举例：交换机与队列一对一完全匹配，如果一个队列绑定到交换机要求routingkey为“dog”，则只转发 routingkey 标记为“dog”的消息，不会转发“dog.puppy”，也不会转发“dog.guard”等等。它是完全匹配、单播的模式。

消息分发：当生产者（P）发送消息时 Rotuing key=black时，这时候将消息传送给 Exchange，Exchange 获取到生产者发送过来消息后，会根据自身的规则进行与匹配相应的 Queue，这时发现 Queue1 和 Queue2 都符合，就会将消息传送给这两个队列。再以其中Queue1 为例，如果Queue1 有多个消费者，默认情况下rabbitMqRabbitMQ会一个一个的发送信息给下一个消费者(consumer)，而不考虑每个任务的时长等等，且是一次性分配，并非一个一个分配。平均的每个消费者将会获得相等数量的消息。这样分发消息的方式叫做round-robin。这可以表示为下图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 4]

### 2.3.2 广播模式 fanout ###

![在这里插入图片描述][20201111141857484.png_pic_center]  
使用场景：它是广播路由消息的最佳选择！

fanout exchange路由消息到所有的与其绑定的queue中，忽略routing key。如果N个queue被绑定到一个fanout exchange，当一条新消息被发布到exchange时，消息会被复制并且传送到这N个queue。

**使用场景：**

1.  大量的多用户在线（multi-player online MMO）游戏使用它更新排行榜或者其他的全体事件
2.  体育新闻网站使用fanout exchange向手机客户端实时发送比分更新
3.  分布式系统可以广播各种状态与配置更新
4.  群聊可以使用fanout exchange让消息在参与者之间传输

### 2.3.3 模糊匹配模式 topic ###

![在这里插入图片描述][20201111142314452.png_pic_center]  
使用场景：它是多路广播路由消息的最佳选择！

topic 交换器通过模式匹配分配消息的路由键属性，将routingkey 和某个模式进行匹配，此时队列需要绑定到一个模式上。它将路由键和绑定键的字符串切分成单词，这些单词之间用点隔开。它同样也会识别两个通配符：符号“\#”和符号“\*”。\#匹配0个或多个单词，\*匹配不多不少一个单词。

## 2.4 ACK应答机制 ##

消费者应用（Consumer applications） - 用来接受和处理消息的应用 - 在处理消息的时候偶尔会失败或者有时会直接崩溃掉。而且网络原因也有可能引起各种问题。这就给我们出了个难题，AMQP代理在什么时候删除消息才是正确的？AMQP 0-9-1 规范给我们两种建议：

1、当消息代理（broker）将消息发送给应用后立即删除。（使用AMQP方法：basic.deliver或basic.get-ok）  
2、待应用（application）发送一个确认回执（acknowledgement）后再删除消息。（使用AMQP方法：basic.ack）

前者被称作自动确认模式（automatic acknowledgement model），后者被称作显式确认模式（explicit acknowledgement model）。在显式模式下，由消费者应用来选择什么时候发送确认回执（acknowledgement）。应用可以在收到消息后立即发送，或将未处理的消息存储后发送，或等到消息被处理完毕后再发送确认回执（例如，成功获取一个网页内容并将其存储之后）。

如果一个消费者在尚未发送确认回执的情况下挂掉了，那AMQP代理会将消息重新投递给另一个消费者。如果当时没有可用的消费者了，消息代理会死等下一个注册到此队列的消费者，然后再次尝试投递。

[RabbitMQ中文文档][RabbitMQ]  
[rmq应答机制1][rmq_1]  
[rmq应答机制2][rmq_2]

## 2.5 TTL过期时间 ##

设置过期时间的两种方式：

> 1、通过设置队列属性，队列中所有的消息都有相同的过期时间  
> 2、对消息本身进行设置，每条消息的ttl时间可以不同

如果两种方法一起使用，则消息的TTL时间以较小值为准。  
消息在队列中的生存时间一旦超过TTL，则会变成死信。  
如果不设置TTL时间，则表示此消息不会过期；如果TTL=0，则表示除非此时可以直接将消息投递给消费者，否则该消息会被立即丢弃。

## 2.6 死信队列 DLX ##

Dead-Letter-Exchange，DLX

当消息在一个队列中变成死信之后，他将会被重新发送到另一个交换器中，这个交换器就是DLX，绑定DLX的队列就是死信队列。

### 2.6.1 什么是死信队列 ###

先从概念解释上搞清楚这个定义，死信，顾名思义就是\`无法被消费的消息 。  
一般来说，producer将消息投递到broker或者直接到queue里了，consumer从queue取出消息进行消费，但某些时候由于特定的原因导致queue中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信，自然就有了死信队列。  
DLX也是一个正常的交换器，和一般交换器没有任何区别，他能在队列上呗指定，实际就是设置某个队列的属性。当这个队列中存在死信时，rmq会自动将这个消息重新发送到设置的DLX上去，进而被路由到另一个队列，即死信队列。可以监听到这个队列中的消息进行相应的处理。

### 2.6.2 RabbitMQ的死信队列 ###

对rabbitmq来说，产生死信的来源大致有如下几种：

1、消息被拒绝（basic.reject或basic.nack）并且requeue=false.
    2、消息TTL过期
    3、队列达到最大长度（队列满了，无法再添加数据到mq中）

“死信”消息会被RabbitMQ进行特殊处理，如果配置了死信队列信息，那么该消息将会被丢进死信队列中，如果没有配置，则该消息将会被丢弃。

### 2.6.3 死信的处理方式 ###

死信的产生既然不可避免，那么就需要从实际的业务角度和场景出发，对这些死信进行后续的处理，常见的处理方式大致有下面几种，

1.  **丢弃**，如果不是很重要，可以选择丢弃
2.  **记录死信入库**，然后做后续的业务分析或处理
3.  **由负责监听死信的应用程序进行处理**

### 2.6.4 应用场景 ###

1、 **过期未支付订单自动取消**  
2、 **每天定时拉取数据**

实际工作中，死信队列可以应用在许多场景中，例如常见的过期未支付订单自动取消 就可以通过 （死信队列 + 过期时间）来实现，就是当有一个队列 queue1，其 对应的死信交换机 为 deadEx1，deadEx1 绑定了一个队列 deadQueue1，  
当队列 queue1 中有一条消息因过期（假设30分钟未支付就取消订单）或者其他原因成为死信的试试，消息就会被转发到死信队列上面，然后我们可以通过监听死信队列中的消息，同时可以加上判断订单的状态是否已经支付，如果已经支付那么不处理，如果未支付，那么可以更新订单状态为已取消。（也就相当于消费的是因过期产生的死信订单信息）。

对比未使用消息队列的时候的解决方案：

> 设置一个定时器，每秒轮询数据库查找超出过期时间且未支付的订单，然后修改状态，但是这种方式会占用很多资源

相比较而言，**使用消息队列可以减少对数据库的压力**，在高流量的情况下可以提高系统的响应速度。

### 2.6.5 死信队列配置 ###

**1、首先需要设置死信队列的exchange和queue，然后进行绑定：**

> Exchange: dlx.exchange  
> Queue: dlx.queue  
> RoutingKey: \#

\#表示只要有消息到达了Exchange，那么都会路由到这个queue上  
**2、然后需要有一个监听，去监听这个队列进行处理  
3、然后我们进行正常声明交换机、队列、绑定，只不过我们需要在队列加上一个参数即可**

`arguments.put(" x-dead-letter-exchange"，"dlx.exchange");，`

**4、消费者通过监听死信队列的消息，查询该订单是否支付，如果没有支付，则取消该订单。这样消息在过期、requeue、 队列在达到最大长度时，消息就可以直接路由到死信队列！**

注意：并不是直接声明一个公共的死信队列，然后所以死信消息就自己跑到死信队列里去了。而是为每个需要使用死信的业务队列配置一个死信交换机，这里同一个项目的死信交换机可以共用一个，然后为每个业务队列分配一个单独的路由key。

有了死信交换机和路由key后，接下来，就像配置业务队列一样，配置死信队列，然后绑定在死信交换机上。也就是说，死信队列并不是什么特殊的队列，只不过是绑定在死信交换机上的队列。死信交换机也不是什么特殊的交换机，只不过是用来接受死信的交换机，所以可以为任何类型`【Direct、Fanout、Topic】`。**一般来说，会为每个业务队列分配一个独有的路由key，并对应的配置一个死信队列进行监听，也就是说，一般会为每个重要的业务队列配置一个死信队列**。

[编程指引][Link 2]

## 2.7 延迟队列 ##

延迟消息是指：当消息被发送以后，并不想立刻让消费者拿到消息，而是等待特定时间后，消费者才能拿到这个消息进行消费。

**在RMQ中本身没有直接支持延迟队列的功能，但是可以通过 DLX+TTL模拟出延迟队列的功能**

使用场景：

> 1、过期未支付订单自动取消  
> 2、每天定时拉取数据、遥控设备在指定时间进行工作等等

实际工作中，死信队列、延时队列可以应用在许多场景中，例如常见的过期未支付订单自动取消 就可以通过 （死信队列 + 过期时间）来实现，就是当有一个队列 queue1，其 对应的死信交换机 为 deadEx1，deadEx1 绑定了一个队列 deadQueue1，  
当队列 queue1 中有一条消息因过期（假设30分钟未支付就取消订单）或者其他原因成为死信的试试，消息就会被转发到死信队列上面，然后我们可以通过监听死信队列中的消息，同时可以加上判断订单的状态是否已经支付，如果已经支付那么不处理，如果未支付，那么可以更新订单状态为已取消。（也就相当于消费的是因过期产生的死信订单信息）。

对比未使用消息队列的时候的解决方案：

设置一个定时器，每秒轮询数据库查找超出过期时间且未支付的订单，然后修改状态，但是这种方式会占用很多资源

相比较而言，使用消息队列可以减少对数据库的压力，在高流量的情况下可以提高系统的响应速度。

## 2.8 优先级队列 ##

优先级默认为0，最高为队列设置的最大优先级。  
优先级搞得队列具有优先被消费的特权。但是前提是：入股哦消费者的消费速度>生产者的速度，且broker中没有消息堆积的情况下，对发送的消息设置邮件及也就没有任何意义。因为生产者岗发送完一条消息就被消费者消费了。

Map<String Object> args = new HashMap<>();
    args.put("x-max-priority",10);
    channel.queueDeclare("queue.priority",true,false,false,args);

# 三、使用规范 #

## 3.1 消息生产者 ##

消息生产者发布消息，需指定交换机名、类型、routingkey等关键信息。要求如下：

• 交换机名称以<组件标识>.<交换机类型>为前缀，以“.”或“\_”进行分割，可多次分割附加说明信息，如nms.topic.nmsExchange。  
• 交换机名保证唯一性。  
• 指定交换机类型

## 3.2 消息消费者 ##

组件作为消息消费者接受消息，需指定与交换机绑定的消息队列及绑定关系routingkey，并对指定的队列进行监听，接受并处理消息。

## 3.3 消息数据类型 ##

发送与接收的消息的数据格式为JSON字符串，具体JSON格式按照组件开发规范标准确定。

## 3.4 可视化管理 ##

rabbitmq提供了一个可视化的监控管理页面rabbitMQ Management，在rabbitmq服务启动后默认使用http://localhost:15672访问，简单方便地对rabbitmq进行管理。  
具体查看rabbitMQ Management指导。

# 四、 开发指导 #

RabbitMQ配置交换机和路由键有两种方法

## 4.1 手动绑定连接—以消费者为例） ##

### 依赖 ###

<dependency>
    	    <groupId>org.springframework.cloud</groupId>
    		<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    </dependency>

### 配置类 ###

package com.common.config;
    
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    
    @Configuration
    public class RabbitMQConfig { 
    
        @Value("${direct.exchange.ai.behavior}")
        private String directExchangeAiBehavior;
    
        @Value("${queue.ai.behavior}") //自己命名的队列名，唯一性，用于持久化
        private String queueAiBehavior;
    
        @Value("${routing.key.ai.behavior}")
        private String routingKeyAiBehavior;
    
        @Bean
        public Queue AIBehaviorQueue() { 
            return new Queue(queueAiBehavior);  //如果要设置ttl时间在这里即可设置！！见下
            /** Map<String, Object> props = new HashMap<>(); // 对于该队列中的消息，设置都等待10s props.put("x-message-ttl", 10000); Queue queue = new Queue("q.pay.ttl-waiting", false, false,false, props); return queue; */
        }
    
        @Bean
        DirectExchange AIBehaviorExchange() { 
            return new DirectExchange(directExchangeAiBehavior);
        }
    
        @Bean
        Binding AIBehaviorBinding() { 
            return BindingBuilder.bind(AIBehaviorQueue()).to(AIBehaviorExchange()).with(routingKeyAiBehavior);
        }
    }

### 消费者 ###

package com.rmq.consumer;
    
    
    import com.alibaba.fastjson.JSONObject;
    import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    
    /** * 接收行为分析服务器处理结果 */
    @Component
    public class AIBehaviorConsumer { 
    
        //行为分析服务器的客流数据类型：区域人数
        private static final String EVENT_TYPE_FRAMES_PEOPLE_COUNTING = "framesPeopleCounting";
    
        @Autowired
        private PassengerDensityService passengerDensityService;
    
        @RabbitListener(queues = "${queue.ai.behavior}")
        public void receiveQueue(Message message) { 
            try { 
               //逻辑处理
                String strFromRMQ = new String(message.getBody());
                JSONObject jsonObject = JSONObject.parseObject(strFromRMQ);
                AnalysisResult analysisResult =
                        JSONObject.toJavaObject(jsonObject.getJSONArray("analysisResult").getJSONObject(0),
                                AnalysisResult.class);
                BehaviorAnalysisResult behaviorAnalysisResult = analysisResult.getBehaviorAnalysisResult().get(0);
                String eventType = behaviorAnalysisResult.getBehaviorAttrs().getEventType();
                String happenTime = jsonObject.getString("dateTime");
                 if (EVENT_TYPE_FRAMES_PEOPLE_COUNTING.equals(eventType)) { 
                    passengerDensityService.saveDensityFromAiBehavior2Map(analysisResult,happenTime);
                }
            } catch (Exception e) { 
                throw new ImmediateAcknowledgeAmqpException(PfmwErrorCode.ERR_PF_AI_DATA_ACCESS_ERROR.getMessage(), e);
            }
        }
    }

## 4.2 spring cloud stream ##

spring对amqp和RabbitMQ都进行了很好的封装，**spring cloud提供了spring cloud stream来对接rabbitmq，这里推荐使用spring cloud stream，可以快速实现消息的发送与接收，无需在配置类中对rabbitmq的exchange、queue进行繁琐的配置。**

spring cloud stream中定义了**binder**的抽象概念，轻松连接rabbitmq，可以通过配置项动态的改变消息的destinations（对应Rabbit MQ 的exchanges）。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 5]

### maven依赖 ###

<dependency> 
                <groupId>org.springframework.cloud</groupId> 
                <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
                <version>1.3.3.RELEASE</version> 
       </dependency>

### application.properties配置工厂的连接信息 ###

#基本配置 
    spring.rabbitmq.port=5672 
    spring.rabbitmq.username=guest 
    spring.rabbitmq.password=guest 
    spring.rabbitmq.host=localhost 
    #更多扩展配置 
    #...

以上是rabbbitmq的基本连接配置（用于连接rmq），除此之外，其余spring cloud stream的配置需要根据生产者和消费者各自的接口进行相应配置，参考实例见下：

### eg1：消息生产者 ###

Spring Cloud Stream 需要定义一个接口来发送消息，如下是一个内置接口，此接口声明了一个名为output的输出binding。

public interface OutputInterface {  
        String OUTPUT = "outputName"; 
     
        @Output(OUTPUT) 
        MessageChannel output(); 
    }

#### application.properties配置生产者的绑定信息 ####

#指定交换机
    spring.cloud.stream.bindings.<outputName>.destination=<destinationName> 
    
    #outputName对应接口中定义的输出通道名字，如output 
    spring.cloud.stream.bindings.<outputName>.destination=nms.direct.nmsExchange 
    
    #配置routingkey，按需配置，不指定默认是#  
    spring.cloud.stream.rabbit.bindings.<outputName>.producer.routing-key-expression='''mykey''' 
    
    #配置交换机类型，按需配置，不指定默认是topic 
    spring.cloud.stream.rabbit.bindings.<outputName>.producer.exchangeType=direct

#### 生产者实例 ####

@EnableBinding(OutputInterface .class)
    public class Sender { 
        @Resource
        OutputInterface  outputInterface ;//对应注解中的接口类
    
        public void send(DeviceMessage msg){ 
            JSONObject json = new JSONObject();
            json.put("message",msg);
            Message<JSONObject> message = MessageBuilder.withPayload(json).build();
            
            /这里的output是接口中的方法，通过指定不同的方法（通道），进行发送!!!!!!!!!!!!!!!!!!!!!!!!
            
            outputInterface.output().send(message);
        }
    }

#### 发送消息 ####

@Autowired
        private Sender sender;
    	//...
        sender.send(deviceMessage);

### eg2： 消费者 ###

Spring Cloud Stream 需要定义一个接口来接收消息，如下是内置的一个接口，此接口声明了一个名为input的输入binding。

public interface InputInterface { 
        String INPUT = "inputName";
    
        @Input( INPUT )
        SubscribableChannel input();
    }

#### application.properties配置消费者的绑定信息 ####

#spring.cloud.stream.bindings.<inputName>.destination=<destinationName> 
    
    #inputName对应接口中定义的输出通道名字，如input 
    spring.cloud.stream.bindings.<inputName>.destination=nms.direct.nmsExchange 
    
    #消息内置转换格式为json 
    spring.cloud.stream.bindings.<inputName>.content-type=application/json 
    
    #分组<groupName>，队列持久化 
    spring.cloud.stream.bindings.<inputName>.group=groupName（自己定义）
     
    #手动ack，每个binding都需要单独设置 
    spring.cloud.stream.rabbit.bindings.<inputName>.consumer.acknowledge-mode=manual 
    
    #配置routingkey，按需配置，不指定默认是#   
    spring.cloud.stream.rabbit.bindings.<inputName>.consumer.bindingRoutingKey=mykey 
    
    #配置交换机类型，按需配置，不指定默认是topic 
    spring.cloud.stream.rabbit.bindings.<inputName>.consumer.exchangeType=direct
    
    #配置队列中消息的过期时间为一分钟
    spring.cloud.stream.rabbit.bindings.<inputName>.consumer.ttl=60000
    
    #配置死信队列中消息的过期时间为一分钟。当超过配置时间之后，该消息会自动的从DLQ队列中移除
    spring.cloud.stream.rabbit.bindings.<inputName>.consumer.dlq-ttl=60000
    
    #开启DLQ（死信队列）
    spring.cloud.stream.rabbit.bindings.<inputName>.consumer.auto-bind-dlq=true
    
    #concurrency并发收消息
    spring.cloud.stream.bindings.<inputName>.consumer.concurrency=3
    
    #当需要使用事务通道的时候，通过如下配置,默认情况下为false
    spring.cloud.stream.rabbit.bindings.<inputName>.consumer. transacted =true
    
    #异常尝试机制，通过ConsumerProperties类中声明，默认3次
    spring.cloud.stream.bindings.<inputName>.consumer.max-attempts=2

**配置说明**

1.指定消息接收的destination，即发送消息的exchange，注意消费者必须确保该exchange正确，否则启动程序后rmq也会自动创建一个错误的持久化交换机。  
2.指定消息内置转换格式为json。  
3.设置分组与持久化。 Spring Cloud Stream创建的**交换机默认是持久化的**，而**队列则是临时队列**，对binding**设置了分组后在rabbitmq中创建的队列将会是持久化队列**，**设置分组以后，可以防止消息被多个实例重复消费。**  
4.设置消息确认模式为手动ack，可按需进行业务处理。  
5.配置routingkey和交换机类型。

#### 消费者实例 ####

@EnableBinding(InputInterface.class) //对应配置中binding所在的接口 
    public class Receiver {  
        private static Logger log = LoggerFactory.getLogger(Receiver.class); 
     
        @StreamListener(InputInterface.INPUT) //对应binding 
        public void receive(Message<JSONObject> message, 
                            @Header(AmqpHeaders.CHANNEL) Channel channel, 
                            @Header(AmqpHeaders.DELIVERY_TAG) Long deliveryTag){  
             
            try {  
                log.info("received:"+message.getPayload()); 
                log.info("channel:"+channel); 
                log.info("deliveryTag:"+deliveryTag); 
                channel.basicAck(deliveryTag, false);//手动确认 
            } catch (IOException e) {  
                e.printStackTrace(); 
            } 
        } 
    }

### 生产者 vs 消费者区别 ###

1.  `生产者`使用`@Output`注解，返回`MessageChannel`
2.  `消费者`使用`@Input`注解，返回`SubscribableChannel`
3.  在消费者/生产者的实例类上都要添加 **@EnableBinding(XXXX.class)**。作用是`@EnableBinding注解接收的参数就是使用@Input或者@Output注解声明了通道（channel）的接口`。Spring Cloud Stream会自动实现这些接口。
4.  @Input和@Output注解的方法有相应的返回值，这些返回值就是对应的通道（channel）对象。要使用通道（channel）时，就只要获取到Spring Cloud Stream对这些接口的实现
5.  **@StreamListener接收的参数是要处理的通道（channel）的名**，所注解的方法就是处理从通道获取到的数据的方法。方法的参数就是获取到的数据。同时，@EnableBinding与@StreamListener注解中的内容需要与自定义接口互相对应。
6.  @Header，@Headers注解获取消息的Header

//消费者接收header
    @StreamListener(target=Sink.INPUT)
    	public void handler1(Message message,@Header(name="contentType") Object header) {
    		System.out.print("狗子收到message消息：" + message.getMessage());
    		System.out.print("消息header：" + header);
    	}
    	
    //生产者发送header
    void sendToMqtt(String data, @Header(MqttHeaders.TOPIC) String topic);

> @Header 注入消息头的单个属性  
> @Headers 注入所有消息头到一个Map中

1.  定义Channel的接口中既可以同时定义 binding 为 “input” 的输入流和 “output” 的输出流。

### 踩坑 ###

**如果收到的消息全是数字，查看接收类是否在正确。  
最好不用String接收。用JsonObject 或者 Message**

### 其他配置项
    下面列出了一些spring cloud stream中rabbitmq的一些其他配置说明，可按需配置，更具体地请参考官方文档
    
    ```java
    #配置DLX死信队列,当这个队列中有死信时，RabbitMQ就会自动的将这个消息重新发布到设置的Exchange上去，进而被路由到另一个队列，可以监听这个队列中消息做相应的处理。 
    
    
    spring.cloud.stream.rabbit.bindings.[input].consumer.autoBindDlq=true 
    spring.cloud.stream.rabbit.bindings.[input].consumer.republishToDlq=true

### 注意点 ###

1.在没有指定交换机类型及routingkey的时候，`spring cloud stream创建的交换机类型默认为topic类型，routingkey为#，交换机名称为destination名称`

2.在不设置分组的情况下，spring cloud stream会自动创建临时队列，程序与rmq连接断开后队列消失。设置分组以后，对应队列就是持久化，且命名为.，例destination=nms.nmsExchange，group=g1，则queue名称为nms.nmsExchange.g1。值得注意的是，设置分组以后消息只会被一个消费者实例消费，防止了消息的重复消费，所以什么时候设置分组应具体情况具体分析。

3.生产者指定消息发送到的exchange的名称、类型以及routingkey，**不需要指定具体队列。**

4.消费者指定消息接收自哪个exchange，消息确认方式，分组（确定队列名）和routingkey。

5.使用spring cloud stream时，rabbitmq的exchange与queue的绑定其实是由消费者端的routingkey配置生成的，生产者只配置发送时消息头中带有的routingkey。

6.如果交换机已存在，那么生产者和消费者配置destination的属性是无效的，如exchangeType。

7.**假如已存在持久化队列nmsqueue，绑定交换机nmsExchange，routingkey为mykey，消费者端配置了新routingkey时会更新原有routingkey**

8.自定义接口时，需要添加相应的配置，修改注解内容。

# 五、持久化 #

rabbitmq持久化：

> **持久化是为提高rabbitmq消息的可靠性，防止在异常情况(重启，关闭，宕机)下数据的丢失**

rabbitmq持久化分为三个部分: **交换器、队列、消息**

## 5.1 交换器的持久化 ##

交换器的持久化是通过声明队列时，将 **durable** 参数设置为**true**实现的。**如果交换器不设置持久化，那么rabbitmq服务重启之后，相关的交换器元数据将会丢失，不过消息不会丢失，只是不能将消息发送到这个交换器中了**，建议将交换器设置为持久化

//public TopicExchange(String name, boolean durable, boolean autoDelete)
    public TopicExchange exchange() { 
            return new TopicExchange(EXCHANGE_NAME,true,false);
        }

## 5.2 队列的持久化 ##

队列的持久化是通过声明队列时，将 **durable** 参数设置为**true**实现的。如果队列不设置持久化，那么rabbitmq服务重启之后，相关的`队列元数据将会丢失`，而消息是存储在队列中的，所以队列中的`消息也会被丢失` 。

//public Queue(String name, boolean durable)
    @Bean
        public Queue queue() { 
            //durable：是否将队列持久化 true表示需要持久化 false表示不需要持久化
            return new Queue(QUEUE_NAME, false);
        }

## 5.3 消息的持久化 ##

队列的持久化只能保证其队列本身的元数据不会被丢失，但是不能保证消息不会被丢失。所以消息本身也需要被持久化，可以在投递消息前设置 **AMQP.BasicProperties** 的属性**deliveryMode**设为**2**即可：

使用convertAndSend方式发送消息，消息默认就是持久化的.
    new MessageProperties() --> DEFAULT_DELIVERY_MODE = MessageDeliveryMode.PERSISTENT --> deliveryMode = 2;

## 5.4 spring cloud stream ##

Spring Cloud Stream创建的**交换机默认是持久化的**，而**队列则是临时队列**，对binding**设置了分组后在rabbitmq中创建的队列将会是持久化队列**，**设置分组以后，可以防止消息被多个实例重复消费。**

## 5.5 将交换器、队列和消息都设置持久化之后就能保证数据不会被丢失吗？ ##

当然不是，多个方面：

*  消费者端: 消费者订阅队列将autoAck设置为true,虽然消费者接收到了消息，但是没有来得及处理就`宕机`了，那数据也会丢失，解决方案就是以为`手动确认接收消息`，待处理完消息之后，手动删除消息
 *  在rabbitmq服务端，如果消息正确被发送，但是rabbitmq`没有来得及持久化`，没有将数据写入磁盘，服务异常而导致数据丢失，解决方案，可以通过rabbitmq集群的方式实现消息中间件的高可用性

## 问题排查指导 ##

### 原生命令 ###

windows系统，找到安装后的 RabbitMQ 所在目录下的 sbin 目录，可以看到该目录下有一些以 rabbitmq 开头的可执行文件，在文件夹url栏中输入cmd回车直接进入命令行，linux系统直接在命令行中进入sbin目录，下面将 RabbitMQ 的安装位置以“.”代替  
1.服务启停  
启动：rabbitmq-server -detached  
关闭：rabbitmqctl stop  
重启：rabbitmq-server restart  
2.插件管理  
开启某个插件：rabbitmq-plugins enable \{插件名\}  
关闭某个插件：rabbitmq-plugins disable \{插件名\}  
查看插件信息：rabbitmq-plugins list  
3.用户管理  
添加用户：rabbitmqctl adduser \{username\} \{password\}  
删除用户：rabbitmqctl deleteuser \{username\}  
修改密码：rabbitmqctl changepassword \{username\} \{newpassword\}  
设置用户角色：rabbitmqctl setuser\_tags \{username\} \{tag\}  
4.权限管理  
权限设置：rabbitmqctl set\_permissions \[-p vhostpath\] \{user\} \{conf\} \{write\} \{read\}  
conf:一个正则表达式match哪些配置资源能够被该用户访问。  
write:一个正则表达式match哪些配置资源能够被该用户读。  
read:一个正则表达式match哪些配置资源能够被该用户访问。  
例：rabbitmqctl set\_permissions -p / root “.*” “.*” “.\*”  
5.查询服务器状态信息  
服务器状态：rabbitmqctl status  
connection信息：rabbitmqctl list\_connections  
Channel信息：rabbitmqctl list\_channels  
Binding信息：rabbitmqctl list\_bindings  
Exchange信息： rabbitmqctl list\_exchanges \[-p vhostpath\] \[exchangeinfoitem …\]  
exchangeinfoitem有：name, type, durable, auto\_delete, internal, arguments.  
Queue信息：rabbitmqctl list\_queues \[-p vhostpath\] \[queueinfoitem …\]  
queueinfoitem有： name, durable, autodelete, arguments, messagesready, messages\_unacknowled, messages, consumers, memory.  
例：rabbitmqctl listqueues name messagesready pid slave\_pids

### 自己例子 ###

#行为分析服务器数据接入配置
    #xalarm
    spring.cloud.stream.bindings.iBehaviorAlarmMessageInput.destination=xalarm_aps_exchange_forward_to_component
    spring.cloud.stream.bindings.iBehaviorAlarmMessageInput.contentType=application/json
    spring.cloud.stream.bindings.iBehaviorAlarmMessageInput.group=mbbias.alarm.ibehavior
    spring.cloud.stream.rabbit.bindings.iBehaviorAlarmMessageInput.consumer.bindingRoutingKey=ibehavior
    spring.cloud.stream.rabbit.bindings.iBehaviorAlarmMessageInput.consumer.exchangeType=direct

package com..rabbitmq.test
    ;
    
    import org.springframework.cloud.stream.annotation.Input;
    import org.springframework.messaging.SubscribableChannel;
    
    public interface RmqChannel { 
    
        String AI_BEHAVIOR_ANALYSIS = "iBehaviorAlarmMessageInput";//行为分析
    
        @Input(AI_BEHAVIOR_ANALYSIS)
        SubscribableChannel iBehaviorAlarmMessageInput();
    
    }

package com.rabbitmq.test;
    
    import com.alibaba.fastjson.JSONObject;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.annotation.StreamListener;
    
    /**
     * @Description 接收行为分析服务器处理结果
     */
    @EnableBinding({ RmqChannel .class})
    public class AiBehaviorConsumer { 
        private static final HikGaLogger LOGGER = HikGaLoggerFactory.getLogger(AiBehaviorConsumer.class);
        //密度
        private static final String EVENT_TYPE_FRAMES_PEOPLE_COUNTING = "framesPeopleCounting";
        private static final String ANALYSIS_RESULT = "analysisResult";
        private static final String TRANS_INFO = "transInfo";
    
        @Autowired
        private PassengerDensityInfoService passengerDensityInfoService;
    
    
        @StreamListener(XalarmChannel.AI_BEHAVIOR_ANALYSIS)
        public void receiveQueueFromXalarm(JSONObject message) { 
            try { 
                JSONObject jsonObject1 = message.getJSONObject(TRANS_INFO);
                AnalysisResult analysisResult = JSONObject.toJavaObject(jsonObject1.getJSONArray(ANALYSIS_RESULT)
                        .getJSONObject(0), AnalysisResult.class);
                BehaviorAnalysisResult behaviorAnalysisResult = analysisResult.getBehaviorAnalysisResult().get(0);
                String eventType = behaviorAnalysisResult.getBehaviorAttrs().getEventType();
                String happenTime = analysisResult.getTimeStamp();
                //客流密度
                if (EVENT_TYPE_FRAMES_PEOPLE_COUNTING.equals(eventType)) { 
                    passengerDensityInfoService.receiveDensityDataFromAIBehavior(analysisResult,happenTime);
                }
    
            } catch (Exception e) { 
                LOGGER.errorWithErrorCode(SubwayAndBusErrorCodeEnum.ACCEPT_AI_DATA_ERROR.getErrorCode(),
                        SubwayAndBusErrorCodeEnum.ACCEPT_AI_DATA_ERROR.getEnMsg(), e);
            }
        }
    }

### rabbitMQ Management指导 ###

使用rabbitMQ提供的rabbitMQ Management可以简单轻松地管理rabbitMQ。  
1.访问  
启动rabbitmq服务后访问rabbitmq management 网页版  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 6]

用户名默认为guest，密码为guest，  
首页如图所示：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 7]  
2.Connections  
查看rabbitmq连接信息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 8]  
点击具体连接可查看更详细信息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 9]  
3.Channel  
查看建立在连接基础上的信道信息，通道为线程级  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 10]  
4.Exchanges  
查看交换机信息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 11]  
点击交换机可以进行更多操作，如查看交换机具体属性，添加绑定队列，发布消息，删除交换机  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 12]  
5.Queue  
查看队列信息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 13]  
点击队列可以进行更多操作，如查看队列具体属性，队列中的消息情况，与交换机的绑定关系bindings，添加绑定关系，发布消息，获得消息，清空队列消息，删除队列等

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20230209/cd2f58da08914061b2a19bc88d354f70.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20230209/1248fdac8dc94923bfc37c9533a87cf2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20230209/9d78fc10732645288f8a3dccbe1f0071.png
[Link 1]: https://www.cnblogs.com/bclshuai/p/8607517.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20230209/bb1ba1e406da4d548482e0446cda1f18.png
[20201111141235458.png_pic_center]: /images/20230209/dd5a04efc5424bdea7f0e100db261431.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20230209/fd7f390a48794ee29a7e2a0dc43312a5.png
[20201111141857484.png_pic_center]: /images/20230209/662e2c71cd2b4a4db8a83b8fbfa3ccc4.png
[20201111142314452.png_pic_center]: /images/20230209/1b6840812711401f9509c2f764ddeb8c.png
[RabbitMQ]: http://rabbitmq.mr-ping.com/AMQP/AMQP_0-9-1_Model_Explained.html
[rmq_1]: https://www.cnblogs.com/haixiang/p/10900005.html#return-%E6%B6%88%E6%81%AF%E6%9C%BA%E5%88%B6
[rmq_2]: https://blog.csdn.net/Anumbrella/article/details/81321701?utm_medium=distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-2.channel_param&depth_1-utm_source=distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-2.channel_param
[Link 2]: https://weishihuai.blog.csdn.net/article/details/97268109?utm_medium=distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-2.channel_param&depth_1-utm_source=distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-2.channel_param
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20230209/59ec977c22004829946281b91f9057d7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20230209/1693d17ea9f94c1e8cb026b8af3711e6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20230209/9578e8b2d553436d9f9630734eed94b1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20230209/3f8d9388095d4641bc8c0e76c6fc55b6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 9]: /images/20230209/85ffff9868064aa8b86281d958c7681d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 10]: /images/20230209/5313683198374df28cf84b48c897bf7e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 11]: /images/20230209/482f94493eae400f840de3a481224737.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 12]: /images/20230209/17f88d1f51de435cbfbafa91574fb751.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 13]: https://img-blog.csdnimg.cn/20200715144334469.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 14]: https://img-blog.csdnimg.cn/20200715144403920.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 15]: https://img-blog.csdnimg.cn/20200715144445372.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center 16]: https://img-blog.csdnimg.cn/20200715144520173.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ==,size_16,color_FFFFFF,t_70#pic_center