仿RabbitMQ实现消息队列——

一、项目简介

需求分析

AMQP 特点：

AMQP 模型：

交换机类型

持久化

网络通信

二、服务端模块

1、交换机数据管理

2、队列数据管理

3、绑定数据管理

4、消息数据管理

5、虚拟机数据管理

6、路由匹配管理

7、消费者管理

8、信道管理

9、连接管理

10、服务器模块

三、客户端模块

1、消费者管理

2、信道管理

3、连接管理

4、异步线程池模块

一、项目简介

在实际的后端开发中, 尤其是分布式系统⾥, 跨主机之间使⽤⽣产者消费者模型, 也是⾮常普遍的需求。因此, 我们通常会把阻塞队列封装成⼀个独⽴的服务器程序, 并且赋予其更丰富的功能。这样的服务程序我们就称为消息队列 (Message Queue, MQ)。

其中 RabbitMQ 是⼀个⾮常知名、功能强⼤且⼴泛使⽤的消息队列。本项目就是仿照RabbitMQ模拟实现一个简单的消息队列。

需求分析

• ⽣产者 (Producer)

• 消费者 (Consumer)

• 中间⼈ (Broker)

• 发布 (Publish)

• 订阅 (Subscribe)

我们需要实现的内容包括：

1.broker服务器：消息队列代理服务器

2.消息发布客户端：向服务器发布消息(生产者）

3.消息订阅客户端：从服务器订阅消息(消费者)

我们的消息队列是基于对AMQP协议的理解进行的整合，那么什么是AMQP协议呢：

AMQP (Advanced Message Queuing Protocol) 是一种网络协议，主要用于消息中间件之间的异步通信。AMQP 提供了一种标准的方式来发送和接收消息，使得不同厂商的消息中间件可以互相操作。

AMQP 特点：

开放标准：AMQP 是一个开放标准，这意味着它的规范是公开的，并且任何人都可以实现它。
二进制协议：AMQP 使用二进制编码，这使得它比基于文本的协议更高效。
可靠性：AMQP 设计为确保消息的可靠传输，包括确认机制、事务支持等。
灵活性：AMQP 支持多种消息路由模式，如点对点 (point-to-point) 和发布/订阅 (publish/subscribe)。
互操作性：AMQP 允许不同厂商的消息中间件互相通信，不受客户端或中间件使用的编程语言的影响。

AMQP 模型：

Broker (消息代理)：这是消息中间件的核心组件，负责接收、存储和转发消息。
Exchange (交换器)：Exchange 接收来自生产者的消息，并根据配置规则将消息发送到一个或多个队列。
Queue (队列)：队列是用来存储消息的数据结构，直到消费者取走这些消息。
Binding (绑定)：绑定定义了 Exchange 和 Queue 之间的关系，确定消息如何从 Exchange 到达 Queue。所谓的 Exchange 和 Queue 可以理解成 "多对多" 关系, 和数据库中的 "多对多" ⼀样. 意思是: ⼀个 Exchange 可以绑定多个 Queue (可以向多个 Queue 中转发消息) ⼀个 Queue 也可以被多个 Exchange 绑定 (⼀个 Queue 中的消息可以来⾃于多个 Exchange)
Producer (生产者)：生产者是向消息中间件发送消息的应用程序。
Consumer (消费者)：消费者是从消息中间件接收消息的应用程序。

总结下来就是这样一张图

对于Broker来说，要实现一下核心API来实现消息队列的基本功能

1.创建交换机

2.销毁交换机

3.创建队列

4.销毁队列

5.创建绑定

6.解除绑定

7.发布消息

8.订阅消息

9.取消订阅

10.确认消息

生产者消费者则通过网络发送请求来调用这些API，实现生产者消费者模型

交换机类型

本项目实现三种交换机类型，也是最常见的：

• Direct: ⽣产者发送消息时, 直接指定被该交换机绑定的队列名

• Fanout: ⽣产者发送的消息会被复制到该交换机的所有队列中

• Topic: 绑定队列到交换机上时, 指定⼀个字符串为 bindingKey。发送消息指定⼀个字符串 routingKey。当 routingKey 和 bindingKey 满⾜⼀定的匹配条件的时候, 则把消息投递到指定队列

持久化

交换机，队列，绑定，消息都是需要持久化的，我们需要根据持久化来保证在程序或主机重启时数据不会丢失。在项目中我们使用了Sqlite数据库来进行本地的轻量级存储

网络通信

⽣产者和消费者都是客⼾端程序, Broker 则是作为服务器，通过⽹络进⾏通信。

我们在broker的基础上，再加上建立连接和打开信道的操作，这样可以更好地复用TCP连接，达到长连接的效果，避免频繁的创建关闭TCP连接。

二、服务端模块

1、交换机数据管理

交换机数据管理就是描述了交换机应该有哪些数据

我们可以设置交换机的类型以及消息的基本属性，基本结构如下：

syntax ='proto3';
package bitmq;
enum ExchangeType
{
    UNKNOWTYPE=0;
    DIRECT=1;
    FANOUT=2;
    TOPIC=3;
};

enum DeliverMode
{
    UNKNOWMODE=0;
    UNDURABLE=1;
    DURABLE=2;
};

message BasicProperties
{
    string id=1;
    DeliverMode delivery_mode=2;
    string routing_key=3;
};
message message
{
    message Payload
    {
        BasicProperties properties=1;
        string body=2;
        string valid=3;
    };
    Payload payload=1;
    uint32 offset=2;
    uint32 length=3;
};

1、交换机的名称：也是交换机的唯一标识

2、交换机的类型：决定了消息的转发方式（三种）

每个队列与交换机绑定信息中有binding_key，每条消息中有routing_key

1.直接交换：binding_key与routing_key相同时，将消息放入队列

2.广播交换：交换机绑定的所有队列都放入消息

3.主题交换：根据具体的匹配算法，将符合匹配条件的binding_key和routing_key对应的队列放入消息。

3、持久化标志：决定当前交换机的数据是否需要持久化存储

4、自动删除标志：如果关联该交换机的客户端都退出了，是否需要自动删除交换机。

5、交换机的其他参数，在本项目中未具体使用。

对交换机的管理：

1、创建交换机：如果已存在就直接成功，不存在就创建（后续的一些结构也是如此操作）

//声明交换机
        bool declareExchange(const std::string &name,ExchangeType type,bool durable,bool auto_delete,
        const google::protobuf::Map<std::string,std::string> &args)
        // bool declareExchange(const std::string &name,ExchangeType type,bool durable,bool auto_delete,
        // const std::unordered_map<std::string,std::string> &args)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it=_exchanges.find(name);
            if(it!=_exchanges.end())
            {
                //如果交换机已经存在，那么直接返回，不需要新增
                return true;
            }
            auto exp=std::make_shared<Exchange>(name,type,durable,auto_delete,args);
            if(durable==true)
            {
                //若为持久化，则添加到数据库中
                bool ret=_mapper.insert(exp);
                if(!ret) return ret;
            }
            //添加到交换机表中
            _exchanges.insert(std::make_pair(name,exp));
            return true;
        }

2、删除交换机：每个交换机都会绑定一个或多个队列，所以在删除前也要删除掉所有相关的绑定信息，若交换机为持久化，也要在数据库中删除，后续结构也是类似操作。

//删除交换机
        void deleteExchange(const std::string &name)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it=_exchanges.find(name);
            if(it==_exchanges.end())
            {
                return;
            }
            if(it->second->durable==true) _mapper.remove(name);
            _exchanges.erase(name);
        }

3、获取指定名称交换机

4、获取当前所有交换机数量

2、队列数据管理

队列数据管理需要管理的数据

1、队列名称

2、持久化存储标志：决定是否将队列持久化存储起来，重启后队列是否依旧存在

3、是否独占标志：如果独占，那么只有当前客户端可以订阅该队列消息

4、自动删除标志：当订阅了当前队列的客户端退出后，是否删除队列。本项目中未实现

5、其他参数

队列管理类（与交换机管理类似）：

1、创建队列

2、删除队列

3、获取指定队列信息

4、获取队列数量

5、获取所有队列的名称：因为系统重启会重新加载数据，消息是以队列为单元存储在文件中的，所以加载消息需要知道队列的名称，因为在存储消息时，存储文件以队列名称进行取名

3、绑定数据管理

绑定数据管理：描述队列与交换机的绑定信息

管理的数据

1、交换机的名称

2、队列的名称

3、对应的binding_key：绑定密钥，在交换机主题交换和直接交换时要用到。

由数字，字符，_,#, . , * 组成，比如news.sport.# 就可以与news.sport.football匹配成功

管理的操作：

1、添加绑定信息

2、解除绑定信息

3、获取交换机所有的相关绑定信息：

1、在删除交换机的时候要删除相关的绑定信息

2、交换机也要通过这些信息来发布到指定的队列

4、获取队列所有的绑定信息：

1、删除队列的时候，也要删除相关的绑定信息

5、获取绑定信息的数量

4、消息数据管理

消息的基本属性

message BasicProperties
{
    string id=1;
    DeliverMode delivery_mode=2;
    string routing_key=3;
};
message message
{
    message Payload
    {
        BasicProperties properties=1;
        string body=2;
        string valid=3;
    };
    Payload payload=1;
    uint32 offset=2;
    uint32 length=3;
};

消息属性：

1、消息ID：唯一标识

2、持久化标志（同队列与交换机）

3、 routing_key：决定了要发布的队列，交换机根据交换类型与binding来匹配

4、消息主体

以下是服务端在管理消息时添加的信息：

存储偏移量：消息以队列为单元存储在文件中，这个偏移量是相对于文件起始位置的偏移量

消息长度：从偏移量位置取出指定长度的消息。解决粘包问题

是否有效标志：标识当前消息是否被删除，因为如果删除消息就进行一次文件读写比较耗费资源，那么我们的策略是（删除一条消息不会将后面的数据拷贝到前边，而是重置了标志位valid,每次删除消息后判断：如果有效消息的数量占消息总数比例不到50%，且数据量超过2000，则进行垃圾回收，重新整理文件系统，当系统重启，也只需要加载有效消息。

消息的管理

以队列为单元进行管理，因为对于消息的所有操作是以队列为单元的

管理数据：

1、消息链表：保存所有待推送的消息

2、待确认hash:消息推送给客户后，会等待客户端确认，收到确认后，才会删除消息。

3、持久化hash：假设消息都需要持久化，操作过程中会垃圾回收，但是垃圾回收会改变存储位置，内存中消息的存储位置也需要改变，要用新位置去更新持久化数据

垃圾回收：将有效消息读出来，然后重新截断文件，将消息写入文件中

4、持久化的有效消息总量

5、持久化的总的消息数量：决定了什么时候进行消息回收

管理操作

1、向队列新增消息

2、获取队首消息：获取消息后，将消息从待推送链表中删除，加入到待确认消息中

3、对消息进行确认：从待确认消息中移除，并进行持久话数据的删除

4、恢复队列的历史消息：主要在构造函数中进行

5、垃圾回收（队列持久化子模块完成）：持久化文件中有效消息比例小于50%，总消息数超过2000进行垃圾回收

6、删除队列相关消息文件：当一个队列被删除了，他的消息也没有存在的意义了。

队列消息管理

1、初始化队列消息结构

2、移除队列消息结构：在队列被删除时调用

3、向队列中新增消息

4、对队列消息进行确认

5、恢复队列历史消息

5、虚拟机数据管理

虚拟机数据管理

对于交换机+队列+绑定信息+消息数据管理的整合

要管理的数据

1、交换机的管理句柄

2、队列数据的管理句柄

3、绑定信息的数据管理句柄

4、消息数据管理句柄

要管理的操作：

1、声明/删除交换机：在删除交换机的时候要删除相关的绑定信息

2、声明/删除队列：在删除队列的时候要删除相关的绑定信息以及数据

3、队列的绑定/解除绑定：绑定的时候，交换机和队列必须存在

4、获取指定队列的消息

5、对指定队列的指定消息进行确认

6、获取交换机相关的所有绑定信息：一条消息要发布给指定交换机的时候，交换机获取所有的绑定信息，来确认消息要发送到哪个队列

6、路由匹配管理

决定了一个消息能否发布到指定的队列中

在交换机与队列的绑定信息中有一个banding_key，这是队列发布的匹配规则

在每条要发布的消息中，都有一个routing_key，这是消息的发布规则

根据交换机的类型进行相关的匹配操作

路由匹配模块本质上来说没有管理的数据，只有向外提供的路由匹配操作：

1、判断routing_key是否符合规定：

格式判定：只能由数字，字母，_, . 构成

2、判断binding_key是否符合规定：

格式判断：只能由数字，字母，_ , # , * 构成

3、判断routing_key与binding_key能否匹配成功的接口

7、消费者管理

客户端有两种：发布消息，订阅消息

只有订阅了消息的客户端才是一个消费者

消费者数据存在的意义：当指定的客户端有了消息后，需要将消息推送给这个消费者客户端

推送的时候就要找到这个客户端的相关信息———连接

消费者信息

1、消费者标识

2、订阅队列名称：当前队列有消息就会推送给这个客户端，当客户端收到消息，需要对指定队列的消息进行确认

3、自动确认标志：自动确认--推送消息后，直接删除消息不需要额外确认，手动确认——推送消息后需要等到确认回复再去删除消息

4、消费处理回调函数指针：队列有一条消息后，通过哪个函数进行处理（向指定客户端推送消息）

消费者管理

以队列为单元进行管理

每个消费者订阅的都是指定队列的消息，消费者对这个消息进行确认也是以队列进行确认

当队列中有消息了，必然是获取订阅了这个队列的消费者进行推送

队列消费者管理结构

数据信息：消费者链表——保存当前队列的所有消费者信息（RR轮转每次取出下一个消费者进行推送——一条消息只需要被一个客户端进行处理即可

管理操作：

1、新增消费者

2、RR轮转获取一个消费者

3、删除消费者

4、队列消费者数量

5、队列的消费者列表是否为空

消费者管理操作（以队列为单元进行管理）

1、初始化队列消费者结构

2、删除队列消费者结构

3、向指定队列中添加消费者

4、删除指定队列的指定消费者

5、获取指定队列的消费者

8、信道管理

信道是网络通信的一个概念，叫做通信信道

网络通信的时候，必然是通过网络连接来完成的，为了充分利用资源，细化出了信道的概念，对于用户来说，一个通信信道就是进行网络通信的载体，而一个真正的通信连接，可以创建出多个信道

每一个信道对于用户来说是独立的，但是本质的底层使用的是一个通信连接。

因此，信道是用户眼中的一个通道，所以所有的网络通信服务都由信道提供

1、声明/删除交换机

2、声明/删除队列

3、绑定信息的绑定/删除

4、消息的发布/订阅队列消息/取消订阅/队列消息的Ack

信道要管理的数据

1、信道ID

2、信道关联的消费者句柄：当信道关闭时，所有关联的消费者订阅都要取消，相当于删除所有的消费者。

3、信道关联的虚拟机句柄

4、工作线程池句柄：信道进行消息发布到指定队列后，要从指定队列的消费者链表中获取一个消费者，对这条消息进行消费，也就是将这条消息推送给客户端的操作要交给线程池来进行。并非每个信道都有一个线程池，而是整个服务器有一个线程池，大家所有的信道都是同一个线程池进行异步操作而已。

信道的管理