# RabbitMQ架构与实战


<!--more-->


## 1. 简介,概念,基本架构

### 1.1 简介

RabbitMQ，俗称“兔子MQ”（可见其轻巧，敏捷），是目前非常热门的一款开源消息中间件，不管是互联网行业还是传统行业都广泛使用（最早是为了解决电信行业系统之间的可靠通信而设计）。

- 高可靠,易扩展,高可用,功能丰富
- 支持大多数编程语言客户端
- 遵循AMQP协议,自身采用Erlang编写
- 支持MQTT协议

### 1.2 整体逻辑架构

![img_7](https://img.zhaojq.top/20260727092000688.png "img_7")

### 1.3 Exchange类型

- Fanout

  会把所有发送到该交换器的消息路由到所有与该交换器绑定的队列中 ![img_8](https://img.zhaojq.top/20260727092000884.png "img_8")

- Direct

  direct类型的交换器路由规则很简单，它会把消息路由到那些BindingKey和RoutingKey完全匹配的队列中

  ![img_9](https://img.zhaojq.top/20260727092001073.png "img_9")

- Topic

  topic类型的交换器在direct匹配规则上进行了扩展，也是将消息路由到BindingKey和RoutingKey相匹配的队列中，这里的匹配规则稍微不同，它约定：

  BindingKey和RoutingKey一样都是由"."分隔的字符串；BindingKey中可以存在两种特殊字符“*”和“#”，用于模糊匹配，其中"*"用于匹配一个单词，"#"用于匹配多个单词（可以是0个）。

  ![img_10](https://img.zhaojq.top/20260727092001276.png "img_10")

- Headers

  headers类型的交换器不依赖于路由键的匹配规则来路由信息，而是根据发送的消息内容中的headers属性进行匹配。在绑定队列和交换器时指定一组键值对，当发送的消息到交换器时，RabbitMQ会获取到该消息的headers，对比其中的键值对是否完全匹配队列和交换器绑定时指定的键值对，如果匹配，消息就会路由到该队列。**headers类型的交换器性能很差，不实用。**

### 1.4 数据存储

- 存储机制

  持久化消息, 非持久化消息(都会写入磁盘)

  > 持久化消息在到达队列时写入磁盘，同时会内存中保存一份备份，当内存吃紧时，消息从内存中清除。这会提高一定的性能。
  >
  > 非持久化消息一般只存于内存中，当内存压力大时数据刷盘处理，以节省内存空间。

 RabbitMQ存储层包含两个部分：**队列索引**和**消息存储**。

![img_11](https://img.zhaojq.top/20260727092001460.png "img_11")

- 队列索引 rabbit_queue_index 

  索引维护队列的落盘消息的信息，如存储地点、是否已被给消费者接收、是否已被消费者ack等。每个队列都有相对应的索引。

  > 索引使用顺序的段文件来存储，后缀为.idx，文件名从0开始累加，每个段文件中包含固定的segment_entry_count条记录，默认值是16384。每个index从磁盘中读取消息的时候，至少要在内存中维护一个段文件，所以设置 queue_index_embed_msgs_below 值得时候要格外谨慎，一点点增大也可能会引起内存爆炸式增长

- 消息存储 rabbit_msg_store

  消息以键值对的形式存储到文件中，一个虚拟主机上的所有队列使用同一块存储，每个节点只有一个。存储分为持久化存储（msg_store_persistent）和短暂存储（msg_store_transient）。持久化存储的内容在broker重启后不会丢失，短暂存储的内容在broker重启后丢失。

  > store使用文件来存储，后缀为.rdq，经过store处理的所有消息都会以追加的方式写入到该文件中，当该文件的大小超过指定的限制（file_size_limit）后，将会关闭该文件并创建一个新的文件以供新的消息写入。文件名从0开始进行累加。在进行消息的存储时，RabbitMQ会在ETS（ErlangTerm Storage）表中记录消息在文件中的位置映射和文件的相关信息。

  消息（包括消息头、消息体、属性）可以直接存储在index中，也可以存储在store中。最佳的方式是**较小的消息存在index中**，而**较大的消息存在store中**。这个消息大小的界定可以通过**queue_index_embed_msgs_below** 来配置，默认值为**4096B**。当一个消息小于设定的大小阈值时，就可以存储在index中，这样性能上可以得到优化。**一个完整的消息**大小小于这个值，就放到索引中，否则放到持久化消息文件中。

  rabbitmq.conf配置信息:

  ```properties
  ## Size in bytes below which to embed messages in the queue index. 
  ## Related doc guide: https://rabbitmq.com/persistence-conf.html 
  ##
  # queue_index_embed_msgs_below = 4096 
  
  ## You can also set this size in memory units 
  ##
  # queue_index_embed_msgs_below = 4kb 1234567
  ```

  > 读取消息时，先根据消息的ID（msg_id）找到对应存储的文件，如果文件存在并且未被锁住，则直接打开文件，从指定位置读取消息内容。如果文件不存在或者被锁住了，则发送请求由store进行处理。
  >
  > 删除消息时，只是从ETS表删除指定消息的相关信息，同时更新消息对应的存储文件和相关信息。在执行消息删除操作时，并不立即对文件中的消息进行删除，也就是说消息依然在文件中，仅仅是标记为垃圾数据而已。当一个文件中都是垃圾数据时可以将这个文件删除。当检测到前后两个文件中的有效数据可以合并成一个文件，并且所有的垃圾数据的大小和所有文件（至少有3个文件存在的情况下）的数据大小的比值超过设置的阈值garbage_fraction（默认值0.5）时，才会触发垃圾回收，将这两个文件合并，执行合并的两个文件一定是逻辑上相邻的两个文件。合并逻辑：
  >
  > 1.锁定这两个文件
  >
  > 2.先整理前面的文件的有效数据，再整理后面的文件的有效数据
  >
  > 3.将后面文件的有效数据写入到前面的文件中
  >
  > 4.更新消息在ETS表中的记录
  >
  > 5.删除后面文件

- 队列结构

  通常队列由**rabbit_amqqueue_process**和**backing_queue**这两部分组成，

  - rabbit_amqqueue_process负责协议相关的消息处理，即接收生产者发布的消息、向消费者交付消
    息、处理消息的确认（包括生产端的confirm和消费端的ack）等。
  - backing_queue是消息存储的具体形式和引擎，并向rabbit_amqqueue_process提供相关的接口以供调用。

  如果消息投递的目的队列是空的，并且有消费者订阅了这个队列，那么该消息会直接发送给消费者，不会经过队列这一步。当消息无法直接投递给消费者时，需要暂时将消息存入队列，以便重新投递。

  rabbit_variable_queue.erl 源码中定义了RabbitMQ队列的**4种状态**：

  ```
  1. alpha：消息索引和消息内容都存内存，最耗内存，很少消耗CPU
  2. beta：消息索引存内存，消息内存存磁盘
  3. gama：消息索引内存和磁盘都有，消息内容存磁盘
  4. delta：消息索引和内容都存磁盘，基本不消耗内存，消耗更多CPU和I/O操作
  ```

  消息存入队列后，不是固定不变的，它会随着系统的负载在队列中不断流动，消息的状态会不断发生变化

  对于**普通没有设置优先级和镜像**的队列来说，backing_queue的默认实现是rabbit_variable_queue，其内部通过**5个子队列**Q1、Q2、delta、Q3、Q4来体现消息的各个状态。![img_12](https://img.zhaojq.top/20260727092001670.png "img_12")

- 为什么消息的堆积导致性能下降？

  在系统负载较高时，消息若不能很快被消费掉，这些消息就会进入到很深的队列中去，这样会增加处理每个消息的平均开销。因为要花更多的时间和资源处理“堆积”的消息，如此用来处理新流入的消息的能力就会降低，使得后流入的消息又被积压到很深的队列中，继续增大处理每个消息的平均开销，继而情况变得越来越恶化，使得系统的处理能力大大降低。解决方案:

  ```
  1.增加prefetch_count的值，即一次发送多条消息给消费者，加快消息被消费的速度。
  2.采用multipleack，降低处理ack带来的开销
  3.流量控制
  ```

## 2. 安装和配置

RabbitMQ与Erlang的兼容关系见: https://www.rabbitmq.com/which-erlang.html

```sh
# 1.安装socat
yum install socat -y
# 2.安装Erlang
wget https://github.com/rabbitmq/erlang-rpm/releases/download/v23.0.2/erlang-23.0.2-1.el7.x86_64.rpm
rpm -ivh erlang-23.0.2-1.el7.x86_64.rpm
# 3.安装rabbitmq
wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.8.5/rabbitmq-server-3.8.5-1.el7.noarch.rpm
rpm -ivh rabbitmq-server-3.8.5-1.el7.noarch.rpm
# 4.启用rabbitmq的管理插件
rabbitmq-plugins enable rabbitmq_management
# 5.启动
systemctl start rabbitmq-server 
或后台启动
rabbitmq-server -detached
# 6.添加用户
rabbitmqctl add_user root 123456
# 7.给用户添加权限
rabbitmqctl set_permissions root -p / ".*" ".*" ".*"
# 8.给用户设置标签
rabbitmqctl set_user_tags root administrator
# 标签和权限如下
None: 没有访问management插件的权限
management: 可以使用消息协议做任何操作的权限
	1. 可以使用AMQP协议登录的虚拟主机的权限
	2. 查看它们能登录的所有虚拟主机中所有队列、交换器和绑定的权限
	3. 查看和关闭它们自己的通道和连接的权限
	4. 查看它们能访问的虚拟主机中的全局统计信息，包括其他用户的活动
policymaker: 所有management标签可以做的，加上：
	1. 在它们能通过AMQP协议登录的虚拟主机上，查看、创建和删除策略以及虚拟主机参数的权限
monitoring: 所有management能做的，加上：
	1. 列出所有的虚拟主机，包括列出不能使用消息协议访问的虚拟主机的权限
	2. 查看其他用户连接和通道的权限
	3. 查看节点级别的数据如内存使用和集群的权限
	4. 查看真正的全局所有虚拟主机统计数据的权限
administrator: 所有policymaker和monitoring能做的，加上：
	1. 创建删除虚拟主机的权限
	2. 查看、创建和删除用户的权限
	3. 查看、创建和删除权限的权限
	4. 关闭其他用户连接的权限
# 9.使用浏览器访问: http://ip:15672 并使用刚设置的用户名密码登录
```

## 3. 常用操作命令

```sh
# 前台启动Erlang VM和RabbitMQ 
rabbitmq-server

# 后台启动 
rabbitmq-server -detached 

# 停止RabbitMQ和Erlang VM 
rabbitmqctl stop 

# 查看所有队列 
rabbitmqctl list_queues 

# 查看所有虚拟主机 
rabbitmqctl list_vhosts

# 在Erlang VM运行的情况下启动RabbitMQ应用 
rabbitmqctl start_app 
rabbitmqctl stop_app 

# 查看节点状态 
rabbitmqctl status 

# 查看所有可用的插件 
rabbitmq-plugins list 

# 启用插件 
rabbitmq-plugins enable <plugin-name>

# 停用插件
rabbitmq-plugins disable <plugin-name>

# 添加用户 
rabbitmqctl add_user username password 

# 列出所有用户： 
rabbitmqctl list_users 

# 删除用户： 
rabbitmqctl delete_user username 

# 清除用户权限： 
rabbitmqctl clear_permissions -p vhostpath username 

# 列出用户权限： 
rabbitmqctl list_user_permissions username 

# 修改密码： 
rabbitmqctl change_password username newpassword

# 设置用户权限： 
rabbitmqctl set_permissions -p vhostpath username ".*" ".*" ".*" 

# 创建虚拟主机: 
rabbitmqctl add_vhost vhostpath

# 列出所以虚拟主机: 
rabbitmqctl list_vhosts 

# 列出虚拟主机上的所有权限: 
rabbitmqctl list_permissions -p vhostpath 

# 删除虚拟主机: 
rabbitmqctl delete_vhost vhost vhostpath

# 移除所有数据，要在 rabbitmqctl stop_app 之后使用: 
rabbitmqctl reset
```

## 4. RabbitMQ工作流程详解

### 4.1 案例

示例代码: **demo02_rabbitmq**(https://gitee.com/ixinglan/rabbitmq-demo.git)

### 4.2 生产者发送消息过程

```
1.生产者连接RabbitMQ，建立TCP连接(Connection)，开启信道（Channel）
2.生产者声明一个Exchange（交换器），并设置相关属性，比如交换器类型、是否持久化等
3.生产者声明一个队列井设置相关属性，比如是否排他、是否持久化、是否自动删除等
4. 生产者通过 bindingKey （绑定Key）将交换器和队列绑定（ binding ）起来
5. 生产者发送消息至RabbitMQ Broker，其中包含 routingKey （路由键）、交换器等信息
6. 相应的交换器根据接收到的 routingKey 查找相匹配的队列。
7. 如果找到，则将从生产者发送过来的消息存入相应的队列中。
8. 如果没有找到，则根据生产者配置的属性选择丢弃还是回退给生产者
9. 关闭信道。
10. 关闭连接。
```

### 4.3 消费者接收消息的过程

```
1. 消费者连接到RabbitMQ Broker ，建立一个连接(Connection ) ，开启一个信道(Channel) 。 
2. 消费者向RabbitMQ Broker 请求消费相应队列中的消息，可能会设置相应的回调函数， 以及
做一些准备工作
3. 等待RabbitMQ Broker 回应并投递相应队列中的消息， 消费者接收消息。
4. 消费者确认( ack) 接收到的消息。
5. RabbitMQ 从队列中删除相应己经被确认的消息。
6. 关闭信道。
7. 关闭连接。
```

### 4.4 Connection和Channel关系

生产者和消费者，需要与RabbitMQBroker建立TCP连接，也就是Connection。一旦TCP连接建立起来，客户端紧接着创建一个AMQP信道（Channel），每个信道都会被指派一个唯一的ID。信道是建立在Connection之上的虚拟连接，RabbitMQ处理的每条AMQP指令都是通过信道完成的。

![img_13](https://img.zhaojq.top/20260727092001832.png "img_13")

> 为什么不直接使用TCP连接，而是使用信道？
>
> RabbitMQ采用类似NIO的做法，复用TCP连接，减少性能开销，便于管理。当每个信道的流量不是很大时，复用单一的Connection可以在产生性能瓶颈的情况下有效地节省TCP连接资源。
>
> 当信道本身的流量很大时，一个Connection就会产生性能瓶颈，流量被限制。需要建立多个Connection，分摊信道。

## 5. RabbitMQ工作模式详解

### 5.1 Work Queue

生产者发消息，启动多个消费者实例来消费消息，每个消费者仅消费部分信息，可达到负载均衡的效果。![img_14](https://img.zhaojq.top/20260727092002013.png "img_14")

示例代码: **demo03_workqueue**(https://gitee.com/ixinglan/rabbitmq-demo.git)

### 5.2 发布/订阅模式

使用fanout类型交换器，**routingKey忽略**。每个消费者定义生成一个队列并绑定到同一个Exchange，每个消费者都可以消费到完整的消息

在RabbitMQ中，生产者不是将消息直接发送给消息队列，实际上生产者根本不知道一个消息被发送到哪个队列

![img_15](https://img.zhaojq.top/20260727092002184.png "img_15")

> 未命名交换器: mq默认交换器, 如果不指定交换器, 就会使用此交换器

临时队列: 发布订阅模式下, 我们使用随机生成的队列名称 `String queueName = channel.queueDeclare().getQueue();`

示例代码: **demo04_fanout**(https://gitee.com/ixinglan/rabbitmq-demo.git)

使用默认交换器, 示例代码: **demo05_default_ex**(https://gitee.com/ixinglan/rabbitmq-demo.git)

### 5.3 路由模式

如果我们想让接收者只接收部分消息，如，我们通过直接模式的交换器将关键的错误信息记录到log文件，同时在控制台正常打印所有的日志信息。

我们可以通过不同的 routingKey 将不同类型的消息发送到不同的队列

> 注意: 如果生产者声明不同队列, 如果消费都没有, 则某类型的消息可能没有消费者, 所以应该让消费者声明队列

例: ![img_16](https://img.zhaojq.top/20260727092002405.png "img_16")

示例代码: **demo06_routingmode**(https://gitee.com/ixinglan/rabbitmq-demo.git)

### 5.4 主题模式

使用 topic 类型的交换器，队列绑定到交换器、 bindingKey 时使用通配符，交换器将消息路由转发到具体队列时会根据消息 routingKey 模糊匹配，比较灵活。

要想 topic 类型的交换器， routingKey 就不能随便写了，它必须得是点分单词。单词可以随便写，生产中一般使用消息的特征。如：“stock.usd.nyse”，“nyse.vmw”，“quick.orange.rabbit”等。**该点分单词字符串最长255字节**。

bindingKey 也必须是这种形式。 topic 类型的交换器背后原理跟 direct 类型的类似：只要队列的 bindingKey 的值与消息的 routingKey 匹配，队列就可以收到该消息。有两个不同：

- `*`匹配一个单词
- `#`匹配0到多个单词

例: ![img_17](https://img.zhaojq.top/20260727092002579.png "img_17")

> 上图中，我们发送描述动物的消息。消息发送的时候指定的 routingKey 包含了三个词，两个点。第一个单词表示动物的速度，第二个是颜色，第三个是物种：<speed>.<color>.<species>。 
>
> 创建三个绑定：Q1绑定到` *.orange.* `, Q2绑定到` *.*.rabbit `和` lazy.# `。如果不能匹配，就丢弃消息
>
> 如果发送的消息 routingKey 是" lazy.orange.male.rabbit "，则会匹配最后一个绑定。
>
> 如果在 topic 类型的交换器中 bindingKey 使用 # ，则就是 fanout 类型交换器的行为
>
> 如果在 topic 类型的交换器中 bindingKey 中不使用 * 和 # ，则就是 direct 类型交换器的行为

示例代码: **demo07_topic**(https://gitee.com/ixinglan/rabbitmq-demo.git)

## 6. Spring整合RabbitMQ

**spring-amqp**是对AMQP的一些概念的一些抽象，**spring-rabbit**是对RabbitMQ操作的封装实现

主要有几个核心类 RabbitAdmin 、 RabbitTemplate 、 SimpleMessageListenerContainer 等 

- RabbitAdmin 类完成对Exchange，Queue，Binding的操作，在容器中管理了 RabbitAdmin 类的时候，可以对Exchange，Queue，Binding进行自动声明。

- RabbitTemplate 类是发送和接收消息的工具类。

- SimpleMessageListenerContainer 是消费消息的容器。

### 6.1 基于xml配置文件的整合

示例代码: **demo08~10**(https://gitee.com/ixinglan/rabbitmq-demo.git)

### 6.2 基于注解的整合

示例代码: **demo11~13**(https://gitee.com/ixinglan/rabbitmq-demo.git)

## 7. SpringBoot整合RabbitMQ

示例代码: **demo14~15**(https://gitee.com/ixinglan/rabbitmq-demo.git)