手搓消息队列【RabbitMQ版】

什么是消息队列？

阻塞队列（Blocking Queue）-> 生产者消费者模型 （是在一个进程内）
所谓的消息队列，就是把阻塞队列这样的数据结构，单独提取成了一个程序，进行独立部署~ --------> 生产者消费模型 （进程和进程之间/服务和服务之间）
生产者消费者模型作用：

• 解耦合
  • 本来有个分布式系统，A服务器 调用 B服务器（A给B发请求，B给A返回响应）===》 A 和 B 的耦合是比较大的！
  • 引入消息队列后，A把请求发送到消息队列，B再从消息队列获取到请求
• 削峰填谷
  • 比如A是入口服务器，A 调用 B 完成一些具体业务，如果是 A 和 B 直接通信，如果突然A 收到一组用户的请求的峰值，此时 B 也会随着受到峰值~
  • 引入消息队列后，A把请求发送到消息队列，B再从消息队列获取到请求。 （虽然A收到很多请求，队列也收到了很多请求，但是B仍旧可以按照原来的节奏处理请求。不至于说一下就收到太多的并发量。）
  • 举个例子：高铁火车站，进站口。 乘客好比A ，进站口好比B，是有限的，就需要一个队列来排队，这样不管人多少，就不会影响到乘客进站以后的坐车。
    

市面上一些知名的消息队列

• RabbitMQ
• Kafka
• RocketMQ
• ActiveMQ
  

需求分析

核心概念1

1. 生产者（Producer）
2. 消费者（Consumer）
3. 中间人（Broker）
4. 发布（Push） 生产者向中间人这里投递消息的过程
5. 订阅（Subscribe） 哪些消费者要从中间人取数据，这个注册的过程，称为 “订阅”
6. 消费 （Consume） 消费者从中间人这里取数据的动作

一个生产者，一个消费者

N个生产者，N个消费者

核心概念2

Broker server 内部也涉及一些关键概念（是为了如何进出队列）

• 虚拟主机（Virtual Host），类似于 MySQL 中的 database，算是一个 “逻辑” 上的数据集合。
  • 一个Broker server 上可以组织多种不同类别数据，可以使用 Virtual Host 做出逻辑上的区分
  • 实际开发中，一个 Broker server也可能同时用来管理多个 业务线上的数据，就可以使用 Virtual Host 做出逻辑上的区分。
• 交换机（Exchange）
  • 生产者把消息投递给 Broker Server，实际上是把消息先交给了 （公司某一层楼）Broker Server 上的交换机，再由交换机把消息交给对应的队列。 （交换机类似于“前台小姐姐”）
• 队列（Queue）
  • 真正用来存储处理消息的实体，后续消费者也是从对应的队列中取数据
  • 一个大的消息队列中，可以有很多具体的小队列
• 绑定（Binding）
  • 把交换机和队列之间，建立关系。
  • 可以把 交换机 和 队列 视为，数据库中 多对多的关系。可以想象，在 MQ 中，也是有一个这样的中间表，所谓的 “绑定’其实就是中间表中的一项
• 消息（Message）
  • 具体来说，是 服务器A 发给 B 的请求（通过MQ转发）， 服务器B 给 服务器A返回的响应（通过MQ转发）
  • 一个消息，可以视为一个字符串（二进制数据），具体由程序员自定义

核心API

消息队列服务器（Broker Server），要提供的核心API

• 创建队列（queueDeclare）
  • 此处不用 Create这样的术语，原因是Create仅仅是创建；而 Declare 起到的效果是，不存在则创建，存在就啥也不做
• 销毁队列（queueDelete）
• 创建交换机（exchangeDeclare）
• 销毁交换机（exchageDelete）
• 创建绑定（queueBind）
• 解除绑定（queueUnbind）
• 发布消息（basicPublish）
• 订阅消息（basicConsume）
• 确认消息（basicAck）
  • 这个API起到的效果，是可以让消费者显式的告诉 broker server,这个消息我处理完毕了，提高整个系统的可靠性~保证消息处理没有遗漏
  • RabbitMQ 提供了 肯定 和 否定的 确认，此处我们项目就只有 肯定确认
    

交换机类型

交换机在转发消息的时候，有一套转发规则的~
提供了几种不同的 交换机类型 （ExchangType）来描述这里不同的转发规则
Rabbit主要实现了四种交换机类型（也是由 AMQP协议定义的）

• Direct 直接交换机
• Fanout 扇出交换机
• Topic 主题交换机
• Header 消息头交换机

项目中实现了前三种

1. Direct 直接交换机
   1. 生产者发送消息时，会指定一个"目标队列"的名字（此时的 routingKey就是 队列的名字）
   2. 交换机收到后，就看看绑定的队列里面，有没有匹配的队列
   3. 如果有，就转发过去（把消息塞进对应的队列中）
   4. 如果没有，消息直接丢弃
2. Fanout 扇出交换机
   1. 会把消息放到交换机绑定的每个队列
   2. 只要和这个交换机绑定任何队列都会转发消息
3. Topic 主题交换机

有两个关键概念

1. bindingKey：把队列和交换机绑定的时候，指定一个单词（像是一个暗号一样）
2. routingKey：生产者发送消息的时候，也指定一个单词
3. 如果当前 bindingKey 和 routingKey 对上了，就可以把消息转发到对应的队列
4. 上述三种交换机类型，就像QQ群发红包

• 专属红包 ======== 直接交换机
• 发个10块钱红包，大家都能领 10块钱红包 ======== 扇出交换机
• 我发个口令红包，只有输入对应口令才能领导红包 ======== 主题交换机
  

持久化

上述 虚拟机、交换机、队列、绑定、消息，需要存储起来。此时内存和硬盘各存储一份，内存为主，硬盘为辅。

• 交换机、队列、绑定：存储在数据库中
• 消息：存储在文件中

在内存中存储的原因：
对于 MQ 来说，能够高效的转发处理数据，是非常关键的指标！ 因此对于使用内存来组织数据，得到的效率，就比放硬盘要高很多
在硬盘中存储原因：
为了防止内存中数据随着进程重启/主机重启而丢失

网络通信

其他的服务器（生产者/消费者）通过网络，和咱们的 Broker Server 进行交互的。
此处设定，使用 TCP + 自定义的应用层协议 实现 生产者/消费者 和 BrokerServer 之间的交互工作

应用层协议主要工作：就是让客户端可以通过网络，调用 brokerserver 提供的编程接口

因此，客户端这边也要提供上述API，只有服务器是真正干实事的；客户端只是发送/接受响应

虽然调用的客户端的方法，但是实际上好像调用了一个远端服务器的方法一样 (远程调用 RPC)

客户端除了提供上述9个方法之外，还需要提供 4个 额外的方法，支撑其他工作

• 1. 创建 Connection
• 2. 关闭 Connection
  • 此处用的 TCP 连接，一个 Connection 对象，就代表一个 TCP连接
• 3. 创建 Channel
  • 一个Connection 里面包含多个 Channel，每个 Channel 上传输的数据都是互不相干的
  • TCP中，建立/断开一个连接，成本挺高的，因此很多时候不希望频繁建立断开 TCP 连接
  • 所以定义一个 Channel ，不用的时候，销毁 Channel，此处 Channel 是逻辑概念，比 TCP 轻量很多
• 4. 关闭 Channel
     

消息应答模式

1. 自动应答，消费者把这个消息取走了，就算应答了
2. 手动应答，basicAck 方法属于手动应答（消费者需要主动调用这个 API 来进行应答）
   

总结

需要做哪些工作？

1. 需要实现 生产者，消费者，brokerserver 三个部分
2. 针对生产者消费者来说，主要编写的是 客户端和服务器的通信部分，给客户端提供一组 api，让客户端的业务代码来调用，从而通过网络通信的方式远程调用 brokerserver 上的方法
   1. 比如创建交换机，客户端这边只需要提供相关参数即可，然后通过 socket 将 request 传入到网卡中，然后服务器从 网卡中读取 request 解析。然后计算请求得到 response，再通过 socket 写回去网卡。
3. 实现 brokerserver 【重点】
4. 持久化

上述的这些关键数据，在硬盘中怎么存储，啥格式存储，存储在数据库还是文件？
后续服务器重启了，如何读取这些数据，把内存中内容恢复过来?

模块划分

点击查看【processon】

创建核心类

Exchange

MSGQueue

Binding

Message

数据库操作

建表操作

此处考虑的是更轻量的数据库SQLite, 因为一个完整的 SQLite 数据库，只有一个单独的可执行文件（不到1M）

1. 直接在pom.xml文件中引入

        
        <dependency>
            <groupId>org.xerialgroupId>
            <artifactId>sqlite-jdbcartifactId>
            <version>3.42.0.0version>
        dependency>
1
2
3
4
5
6

2. 然后在 application.yml配置文件中

spring:
  datasource:
    url: jdbc:sqlite:./data/meta.db
    username:
    password:
    driver-class-name: org.sqlite.JDBC
1
2
3
4
5
6

上述依赖和配置都弄完后，当程序启动时，会自动建立数据库。所以我们只需要建表就行。
此处我们根据之前的需求分析，建立三张表，此处我们通过 代码形式来建造三张表

1. 配置application.yml

mybatis:
  mapper-locations: classpath:mapper/**Mapper.xml
1
2

2. 创建一个对应的 interface

3. 创建 mapper目录和文件 MetaMapper.xml

交换机操作

1. 在接口先写方法

void insertExchange(Exchange exchange);
List<Exchange> selectAllExchanges();
void deleteExchange(String exchangeName);
1
2
3

2. 在 xml 中写

<insert id="insertExchange" parameterType="com.example.mq.mqserver.core.Exchange">
  insert into exchange values (#{name},#{type},#{durable},#{autoDelete},#{arguments});
insert>

<select id="selectAllExchanges" resultType="com.example.mq.mqserver.core.Exchange">
  select * from exchange;
select>

<delete id="deleteExchange" parameterType="java.lang.String">
  delete from exchange where name = #{exchangeName};
delete>
1
2
3
4
5
6
7
8
9
10
11

队列操作

1. 在接口中先写方法

void insertQueue(MSGQueue queue);
List<MSGQueue> selectAllQueues();
void deleteQueue(String queueName);
1
2
3

2. 在xml中写

<insert id="insertQueue" parameterType="com.example.mq.mqserver.core.MSGQueue">
  insert into queue values (#{name},#{durable},#{exclusive},#{autoDelete},#{arguments});
insert>

<select id="selectAllQueues" resultType="com.example.mq.mqserver.core.MSGQueue">
  select * from queue;
select>

<delete id="deleteQueue" parameterType="java.lang.String">
  delete from queue where name = #{queueName};
delete>
1
2
3
4
5
6
7
8
9
10
11

绑定操作

1. 在接口中先写方法

void insertBinding(Binding binding);
List<Binding> selectAllBindings();
void deleteBinding(Binding binding);
1
2
3

2. 在xml中写

<insert id="insertBinding" parameterType="com.example.mq.mqserver.core.Binding">
  insert into binding values (#{exchangeName},#{queueName},#{bindingKey});
insert>

<select id="selectAllBindings" resultType="com.example.mq.mqserver.core.Binding">
  select * from binding;
select>

<delete id="deleteBinding" parameterType="java.lang.String">
  delete from binding where exchangeName = #{exchangeName} and queueName = #{queueName};
delete>
1
2
3
4
5
6
7
8
9
10
11

一个统一的类进行数据库操作

在服务器（BrokerServer）启动的时候，能够做出以下逻辑判定：

• 1. 如果数据库存在，表也都有了，不做任何操作
• 2. 如果数据库不存在，则创建库，创建表，构造默认数据

构造一个类 DataBaseManager

package com.example.mq.mqserver.datacenter;
import com.example.mq.MqApplication;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.ExchangeType;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.mapper.MetaMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.File;
import java.lang.reflect.Field;
import java.util.List;

/**
 * 通过这个类，来整合数据库操作
 */
public class DataBaseManager {
   
    private MetaMapper metaMapper;
    // 针对数据库进行初始化
    public void init(){
   
        // 要做的是从 Spring 获取到现成的对象
        metaMapper = MqApplication.context.getBean(MetaMapper.class);
        if(!checkDBExists()){
   
            // 数据库不存在,就进行建库建表操作
            // 先创建一个 data 目录
            File dataDir = new File("./data");
            dataDir.mkdirs();
            // 创建数据表
            createTable();
            // 插入默认数据
            createDefaultData();
            System.out.println("[DataBaseManager] 数据库初始化完成！");
        }else {
   
            // 数据库已经存在，则什么都不做
            System.out.println("[DataBaseManager] 数据库已经存在！");
        }
    }
    public void deleteDB(){
   
        File file = new File("./data/meta.db");
        boolean ret = file.delete();
        if (ret){
   
            System.out.println("[DataBaseManager] 删除数据库文件成功！");
        }else {
   
            System.out.println("[DataBaseManager] 删除数据库文件失败！");
        }
        File dataDir = new File("./data");
        ret = dataDir.delete();
        if (ret){
   
            System.out.println("[DataBaseManager] 删除数据库目录成功！");
        }else {
   
            System.out.println("[DataBaseManager] 删除数据库目录失败！");
        }
    }
    private boolean checkDBExists() {
   
        File file = new File("./data/meta.db");
        if (file.exists()){
   
            return true;
        }
        return false;
    }
    // 这个方法用来建表
    // 建库操作并不需要手动执行（不需要手动创建 meta.db 文件）
    // 首次执行这里的数据库操作的时候，就会自动创建 meta.db 文件 （mybatis 帮我们完成的）
    private void createTable() {
   
        metaMapper.createExchangeTable();
        metaMapper.createQueueTable();
        metaMapper.createBindingTable();
        System.out.println("[DataBaseManager] 创建表完成！");
    }

    // 给数据库表中，添加默认的值
    // 此处主要是添加一个默认的交换机
    // RabbitMQ 里有一个这样的设定: 带有一个 匿名 的交换机，类型是 DIRECT
    private void createDefaultData() {
   
        // 构造一个默认交换机
        Exchange exchange = new Exchange();
        exchange.setName("");
        exchange.setType(ExchangeType.DIRECT);
        exchange.setDurable(true);
        exchange.setAutoDelete(false);
        metaMapper.insertExchange(exchange);
        System.out.println("[DataBaseManager] 创建初始数据完成");
    }
    // 把其他的数据库操作，也在这个类封装下
    public void insertExchange(Exchange exchange){
   
        metaMapper.insertExchange(exchange);
    }
    public List<Exchange> selectAllExchanges(){
   
        return metaMapper.selectAllExchanges();
    }
    public void deleteExchange(String exchangeName){
   
        metaMapper.deleteExchange(exchangeName);
    }
    public void insertQueue(MSGQueue queue){
   
        metaMapper.insertQueue(queue);
    }
    public List<MSGQueue> selectAllQueues(){
   
        return metaMapper.selectAllQueues();
    }
    public void deleteQueue(String queueName){
   
        metaMapper.deleteQueue(queueName);
    }
    public void insertBinding(Binding binding){
   
        metaMapper.insertBinding(binding);
    }
    public List<Binding> selectAllBindings(){
   
        return metaMapper.selectAllBindings();
    }
    public void deleteBinding(Binding binding){
   
        metaMapper.deleteBinding(binding);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

消息持久化

消息存储格式

Message，如何在硬盘上存储？

1. 消息操作并不涉及到复杂的增删改查
2. 消息数量可能会非常多，数据库的访问效率并不高

所以要把消息直接存储在文件中
以下设定消息具体如何在文件中存储~

消息是依托于队列的，因此存储的时候，就要把 消息 按照 队列 维度展开
此处已经有了一个 data 目录（meta.db就在这个目录中）
在 data 中创建一些子目录，每个队列对应一个子目录，子目录名就是队列名

queue_data.txt：这个文件里面存储的是二进制的数据，我们约定转发到这个队列的队列所有消息都是以二进制的方式进行存储

首先规定前4个字节代表的该消息的长度，后面紧跟着的是消息本体。
对于BrokerServer来说，消息是需要新增和删除的。
生产者生产一个消息，就是新增一个消息
消费者消费一个消息，就是删除一个消息
对于内存中的消息新增删除就比较容易了：使用一些集合类就行
对于文件中新增：
我们采用追加方式，直接在当前文件末尾新增就行
对于文件中删除：
如果采用真正的删除，效率就会非常低。将文件视为顺序表结构，删除就会涉及到一系列的元素搬运。
所以我们采用逻辑删除的方式。根据消息中的一个变量 isValid 判断该消息是否有效，1 为有效消息；0 为
无效消息

那么如何找到每个消息对应在文件中的位置呢？ 我们之前在 Message 中设置了两个变量，一个是 offsetBeg,一个是 offsetEnd。
我们存储消息的时候，是同时在内存中存一份和硬盘中存一份。而内存中存到那一份消息，记录了当前的消息的 offsetBeg 和 offsetEnd。通过先找到内存中的消息，再根据该消息的两个变量值，就能找到硬盘中的消息数据了。

垃圾回收

随着时间的推移，文件中存放的消息可能会越来越多。并且可能很多消息都是无用的，所以就要针对当前消息数据文件进行垃圾回收。

此处我们采用的复制算法，原理也是比较容易理解的 （复制算法：比较适用的前提是，当前的空间，有效数据不多，大多数都是无效的数据）
直接遍历原有的消息数据文件，把所有的有效数据数据重新拷贝一份到新的文件中，新文件名字和原来文件名字相同，再把旧的文件直接删除掉。

那么垃圾回收的算法有了，何时触发垃圾回收？

此处就要用到我们每个队列目录中，所对应的另一个文件 queue_stat.txt了，使用这个文件来保存消息的统计信息
只存一行数据，用 \t 分割， 左边是 queue_data.txt 中消息的总数目，右边是 queue_data.txt中有效的消息数目。 形如 2000\t1500, 代表该队列总共有2000条消息，其中有效消息为1500条
所以此处我们就约定，当消息总数超过2000条，并且有效消息数目低于总消息数的50%，就处罚一次垃圾回收GC

如果当一个文件消息数目非常的多，而且都是有效信息，此时会导致整个消息的数据文件非常庞大，后续针对这个文件操作就会非常耗时。假设当前文件已经达到10个G了，那么此时如果触发一次GC，整个耗时就会非常高。

对于RabbitMQ来说，解决方案：
文件拆分：当某个文件长度达到一定的阈值的时候，就会拆分成两个文件（拆着拆着就成了很多文件）
文件合并：每个单独的文件都会进行GC，如果GC之后，发现文件变小了，就会和相邻的其他文件合并
这样做，可以保证在消息特别多的时候，也能保证性能上的及时响应
实现思路：

1. 用一个专门的数据结构，来存储当前队列中有多少个数据文件，每个文件大小是多少，消息的数目是多少，无效消息是多少
2. 设计策略：什么时候触发文件拆分，什么时候触发文件合并
   

统计文件读写

需要定义一个内部类，在表示该队列的统计消息，此处优先考虑 static 静态内部类

static public class Stat {
   
    // 此处直接定义成 public
    public int totalCount;  // 总的消息数
    public int validCount; // 有效消息数
}
1
2
3
4
5
6

1. 统计文件的读

private Stat readStat(String queueName) {
   
Stat stat = new Stat();
try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
   
    Scanner scanner = new Scanner(inputStream);
    stat.totalCount = scanner.nextInt();
    stat.validCount = scanner.nextInt();
    return stat;
} catch (IOException e) {
   
    e.printStackTrace();
}
return null;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

2. 统计文件的写

private void writeStat(String queueName, Stat stat) {
   
    // 使用 PrintWrite 来写文件
    // OutputStream 打开文件，默认情况下，会直接把源文件清空，此时就相当于 新数据把旧的数据覆盖了
    // 加个 参数 true，就会变成追加 new FileOutputStream(getQueueStatPath(queueName),true)
    try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
   
        PrintWriter printWriter = new PrintWriter(outputStream);
        printWriter.write(stat.totalCount + "\t" + stat.validCount);
        printWriter.flush();
    } catch (IOException e) {
   
        e.printStackTrace();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

创建消息目录和文件

1. 先创建队列对应的目录（以队列名字为名的目录）
2. 创建队列里面的消息数据文件
3. 创建队列里面的消息统计数据文件
4. 给消息统计文件设置初始值

// 创建队列对应的文件目录
public void createQueueFiles(String queueName) throws IOException {
   
    // 1. 先创建队列对应的消息目录
    File baseDir = new File(getQueueDir(queueName));
    if (!baseDir.exists()) {
   
        // 不存在就创建这个目录
        Boolean ok = baseDir.mkdirs();
        if (!ok) {
   
            throw new IOException("创建目录失败！baseDir=" + baseDir.getAbsolutePath());
        }
    }
    // 2. 创建队列数据文件
    File queueDataFile = new File(getQueueDataPath(queueName));
   
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16