消息队列学习

用户11719958

发布于 2025-12-30 12:04:42

710

一，项目介绍

首先介绍一下阻塞队列（Blocking Queue），在多线程编程中阻塞队列(Blocking Queue)是⼀种常⽤于实现⽣产者和消费者模型的数据结构。

其与普通的队列区别在于，当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放⼊了元素；当队列满时，往队列⾥存放元素的操作也会被阻塞，直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的，线程在对阻塞队列进行操作时会被阻塞)。

而阻塞队列是应用于同一主机的不同线程之上。

所谓的消息队列，就是把阻塞队列这样的数据结构，单独提取成一个程序，进行独立部署。

分布式系统里, 跨主机之间使用生产者消费者模型, 也是非常普遍的需求。

因此, 我们通常会把阻塞队列封装成一个独立的服务器程序, 并且赋予其更丰富的功能。这样的服务程序我们就称为消息队列 (Message Queue, MQ)。也是跨主机实现生产者消费者模型。

生产者消费者模型的作用：

解耦合：使数据的生产与数据的处理分离开来，使程序更加模块化。
支持并发：根据实际的生产速率与消费速率，控制创建不同数量生产者的线程或消费者的线程。
支持忙闲不均/削峰填谷：当生产的数据量剧增时，消息队列可以作为缓冲区，将数据缓存起来，消费者根据自己的处理能力来获取数据。

市面上成熟的消息队列非常多：

RabbitMQ
Kafka
RocketMQ
ActiveMQ
......

其中 RabbitMQ 是一个非常知名、功能强大且广泛使用的消息队列。这个项目就是仿照RabbitMQ，实现一个简单的消息队列。

二，开发环境

操作系统：Linux(Ubuntu 20.04)
编辑器：Vscode/vim
编译器/调试工具：g++/gdb
项目自动化构建工具：Makefile

三，技术选型

开发语言：C++
序列化框架：protobuf序列化框架
网络通信：自定义应用层协议+muduo库
数据持久化：sqlite3数据库
单元测试框架：Gtest

四，第三方库介绍

1，Protobuf

ProtoBuf（全称 Protocol Buffer）是数据结构序列化和反序列化框架，它具有以下特点：

语言无关、平台无关：即 ProtoBuf 支持 Java、C++、Python 等多种语言，支持多个平台。
高效：即比 XML 更小、更快、更为简单。
扩展性、兼容性好：你可以更新数据结构，而不影响和破坏原有的旧程序。

protobuf使用流程介绍：

编写 .proto 文件，目的是为了定义结构对象（message）及属性内容。
使用 protoc 编译器编译 .proto 文件，生成一系列接口代码，存放在新生成头文件和源文件中。
依赖生成的接口，将编译生成的头文件包含进我们的代码中，实现对 .proto 文件中定义的字段进行设置和获取，和对 message 对象进行序列化和反序列化。

示例：

1，编写contacts.proto文件

//通过protobuf生成代码

//声明语法版本
syntax="proto3";

//声明代码的命名空间
package contacts;


//结构化对象的描述
message contact
{
    //各个字段的描述：子段类型 子段名=字段编号
    uint64 sn=1;
    string name=2;
    float score=3;
};

2，编译contacts.proto文件

//编译命令格式
protoc --cpp_out=./ contacts.proto

3，生成的文件

contacts.pb.cc 和contacts.pb.h

4，序列化和反序列的使用

#include <iostream>
#include <string>
#include "contacts.pb.h"

int main()
{
    // 序列化
    contacts::contact conn;
    conn.set_sn(1001);
    conn.set_name("小明");
    conn.set_score(90);
    std::string str = conn.SerializeAsString();
    // std::cout<<str<<std::endl;

    // 反序列化
    contacts::contact stu;
    bool ret = stu.ParseFromString(str);
    if (ret == false)
    {
        std::cout << "反序列化失败" << std::endl;
    }
    std::cout << stu.sn() << std::endl;
    std::cout << stu.score() << std::endl;
    std::cout << stu.name() << std::endl;
    return 0;
}

2，muduo库

2.1，muduo库的介绍

muduo是由陈硕大佬开发，是一个基于非阻塞IO事件驱动的C++高并发TCP网络编程库。它是一个基于主从Reactor模型的网络库，其用的线程模型是one thread one loop，所谓的one thread one loop指的是：

一个线程只能有一个事件循环（EventLoop），用来响应IO事件。
一个文件描述符，只能由一个线程进行读写，也就是一个TCP连接必须归属于某个EventLoop管理。

在搭建服务器和客户端时，会使用到muduo库中的相关接口。

3，SQLite

SQLite是一个较轻量级的数据库，它是一个零配置的数据库，这意味着与其他数据库不一样，我们不需要在系统中配置。

像其他数据库，SQLite 引擎不是一个独立的进程，可以按应用程序需求进行静态或动态连接，SQLite 直接访问其存储文件。

SQLite3 C/C++ API介绍

SQLite3 官方文档：https://www.sqlite.org/c3ref/funclist.html

//sqlite3有 三种安全等级
//1，非线程安全模式：不加任何的锁，此时的效率肯定是最高的
//2，线程安全模式（不同的连接在不同的线程/进程下使用是安全的，即同一个操作句柄不能用于多线程间）
//3，串行化模式（可以在不同的线程/进程间使用同一个句柄）：对所有的操作都 加锁保护起来

//下面的所有函数，返回值为SQLITE_OK表示函数执行成功，否则失败。

//1，创建/打开数据库文件，并返回操作句柄（以输出型参数的形式），该函数执行成功返回SQLITE_OK
int sqlite3_open(const char *filename, sqlite3 **ppDb)

//2，若在编译阶段启动了线程安全，则在程序运行阶段可以通过参数选择线程安全等级
int sqlite3_open_v2(const char *filename, sqlite3 **ppDb, int flags, const char *zVfs );
falg参数可以设置为：SQLITE_OPEN_READWRITE-- 以可读可写方式打开数据库文件
                   SQLITE_OPEN_CREATE-- 不存在数据库文件则创建
                   SQLITE_OPEN_NOMUTEX--多线程模式，只要不同的线程使用不同的连接即可保证线程安全
                   SQLITE_OPEN_FULLMUTEX--串行化模式

//3，执行语句
//对于执行语句sql，他可能是一个查询语句，那么对于结果如何处理？
//此时就需要我们传入一个回调函数来对结果进行处理
//如果这个sql语句是一个插入语句或者删除语句等等，那么回调函数设置为nullptr即可
//在该函数的执行过程中，会将arg传给回调函数的第一个参数
//最后一个参数表示，如果该执行出错，错误信息会设置到err中，通过这个我们可以查看错误原因
int sqlite3_exec(sqlite3*, char *sql, int (*callback)(void*,int,char**,char**),
void* arg, char **err)
callback回调函数的参数：  
    void* : 原函数中的arg参数
    int：一行中数据的列数
    char**：存储一行数据的字符指针数组
    char**：每一列的字段名称   
注意：这个回调函数有一个int类型的返回值，这个回调函数如果成功执行必须返回0，返回非0就会触发abort退出程序
//4，销毁语句
int sqlite3_close(sqlite3* db); 成功返回 SQLITE_OK
int sqlite3_close_v2(sqlite3*); 推荐使用--无论如何都会返回SQLITE_OK

//5，获取错误信息
const char *sqlite3_errmsg(sqlite3* db);

4，GTest

GTest 是一个跨平台的 C++单元测试框架，由 google 公司发布。它提供了丰富的断言、致命和非致命判断、参数化等等。

TEST(test_case_name, test_name)
TEST_F(test_fixture,test_name)

TEST：主要用来创建一个简单测试，它定义了一个测试函数，在这个函数中可以使用任何 C++代码并且使用框架提供的断言进行检查。
TEST_F：主要用来进行多样测试，适用于多个测试场景如果需要相同的数据配置的情况，即相同的数据测不同的行为。

GTest 中的断言的宏可以分为两类：

ASSERT_系列：如果当前点检测失败则退出当前函数。
EXPECT_系列：如果当前点检测失败则继续往下执行。

//断言宏的使用
        ASSERT_GT(age,18) //age是否大于18
        ASSERT_LT(age,18) //age是否小于18
//同理也存在EXPECT_GT EXPECT_LT
        ASSERT_ 断言失败停止运行
        EXPECT_ 断言失败继续运行

示例：

#include <iostream>
#include <gtest/gtest.h>

//第一个参数表示单元测试的总名称，第二个表示具体的名称
TEST(test,great_than)
{
    int age=20;
    //GT大于
    ASSERT_GT(age,18);
    std::cout<<"OK!\n";    
}
TEST(test,less_than1)
{
    int age=20;
    //LT小于
    EXPECT_GT(age,18);
    std::cout<<"OK!\n";
}
int main(int argc,char* argv[])
{
    testing::InitGoogleTest(&argc,argv);
    //启动所有的单元测试
    int ret=RUN_ALL_TESTS();
    return 0;
}

五，项目需求分析

1，核心概念

在前面提到过，消息队列，就是把阻塞队列这样的数据结构，单独提取成一个程序，进行独立部署。

会有很多个生产者客户端向服务器放入数据，服务器在将消息推送给对应的客户端进行处理。

Broker Server服务器内部涉及到一些关键的概念： 1，队列(Queue)：真正用来存储消息的实体，后续消费者也是从队列中获取数据。

2，交换机(Exchange)：生产者将消息投递到Broker Server上，是先将消息船体给某个交换机，再由交换机决定，应该将消息传递给哪些绑定的队列。

3，绑定(Binding)：交换机与队列的绑定关系，一个交换机可以绑定一个或者多个队列。

4，虚拟机(Virtual Host)：虚拟机可以理解为数据库中的"database"。一个Broker Server上可以组织不同类别的数据，此时就可以使用虚拟机进行区分。但是在本项目的实现中，虚拟机只实现了一个。

5，消息(Message)：可以认为是A给B发的请求，通过MQ转发，就是一个消息。同样B给A发出响应，经过MQ转发，也是一个消息。

当生产者发布一条消息到服务器后，先交给交换机，再由交换机决定交给哪个队列，这个过程是由交换机类型决定的。交换机有三种类型：Direct（直接交换），Fanout（广播交换），Topic（主题交换）。

生产者发送消息的时候，会携带一个routing_key，就是一个字符串，而交换机与队列的绑定关系中也包含一个字符串binding_key
Direct（直接交换）：生产者发送消息的时候，交换机先拿到消息，交换机再根据绑定的所有队列的binding_key进行比较，只有routing_key=binding_key，才将消息发送给该队列。
Fanout（广播交换）：同样生产者发送消息先到交换机，此时交换机会将这条消息发送给所有绑定的队列。
Topic（主题交换）：可以理解为如果此时routing_key与binding_key对上"暗号"了，此时就可以把消息发送给指定的队列。这里的"暗号"实际上就是指一种匹配规则，当routing_key与binding_key匹配成功时，就可以把消息发送给指定的队列中。

以上的这些内容，都是AMQP协议——高级消息队列协议中规定出来的。

持久化操作：对于上述的数据：交换机，队列，绑定关系，消息等等，需要在内存中存储，也需要在硬盘中存储。在内存中存储：使用/访问更加方便开快速。在硬盘存储：保证服务器重启后数据不丢失。

至此，本项目需要实现的内容包括：Broker Server服务器，客户端（包括发布客户端和订阅客户端）。

2，服务器需要提供的核心API

创建交换机（exchangeDeclare）
移除交换机（exchangeDelete）
创建队列（queueDeclare）
移除队列（queueDelete）
创建绑定关系（queueBind）
解除绑定关系（queueUnBind）
发布消息（basicPublish）
订阅消息（basicConsume）
确认消息（basicAck）：消费客户端拿到消息，处理完毕后，向服务器发送确认响应，然后就可以从服务器上删除该消息了。
取消订阅（basicCancel）

3，网络通信

生产者/消费者都是通过网络，和Broker Server进行交互的。

此处设定，使用Tcp+自定义应用层协议实现生产者/消费者和Broker Server的交互。

在客户端这边，也需要提供上述服务器的方法，只不过服务器上的方法是真正做事的，而客户端这边实现的方法，实际上只是发送请求/接受响应。

由于在Tcp中，建立/断开一个连接（Connection）的成本其实挺高的，因此，很多时候，不希望频繁的建立/断开Tcp连接。

为此，RabbitMQ引入一个新的概念——信道（channel），一个连接（Connection）可以创建多个信道，信道（Channel）只是逻辑上的一个概念，它比连接（Connection）轻量很多。

每个信道上的数据传输都是互不相干的，这是它的核心价值。它允许在同一个Tcp连接上，并发地执行多个独立的操作（如发布消息，消费消息，声明队列等等），这些操作都是在不同信道上进行的，互不干扰。

Tcp连接和信道的关系就比如高速公路和车道的关系，Tcp是铺路，信道是划车道。

如上图所示，在发布客户端中，不同的线程使用不同的信道来发送请求，这几个信道是互不影响的，一个信道上操作失败通常是不会影响其他信道。在服务器这边，也有信道，来接受客户端发送来的请求，并进行处理（这么多的请求，到时候肯定会选择使用线程池来处理）。

在这里我有一个问题，就是为什么客户端也要使用信道这个概念，发送请求时，为什么不直接使用多线程，而是要使用信道。也就是下面的这种方式：

如果使用这种方案：会存在以下问题。

因为对于客户端发送的一些请求来说，是有严格顺序的，比如客户端发送了声明队列和订阅队列消息这两个请求，而服务器可能先接受到的请求是订阅队列消息，但是此时队列根本还没有创建，就无法订阅，程序混乱。
当服务端处理完客户端发送的请求后，会给客户端发送一个响应，如果使用这种方式，我们无法知道这个响应是哪个线程对应的请求。

4，消息应答模式

被消费的消息，需要进行应答，有两种应答方式：

自动应答：消息只要消费了，就算应答完毕了，Broker Server直接删除掉这条消息。

手动应答：消费者消费完消息后，需要收到调用接口，像服务器发送响应，然后Broker Server再删除掉这条消息。

六，项目模块化划分

有了上面对消息队列的简单认识，接下来就要细化一下具体要做哪些工作。

需要实现的内容包括：

服务器（Broker Server）发布客户端（生产者）订阅客户端（消费者）

1，服务端模块设计

交换机模块
队列模块
绑定关系模块
消息模块
虚拟机模块：该模块是对以上四个模块的整合，向外提供操作接口。
路由匹配模块：根据交换机的类型，判断一条消息应该路由给绑定的哪些队列。
消费者管理模块：消费者就是订阅者，当消费者调用API订阅了某条消息，其实本质是消费者订阅了某个队列。然后当这个队列有消息时，就会查找这个队列有哪些消费者订阅了，将消息推送给其中一个消费者进行处理，所以在服务器这边，我们需要将消费者的消息保存起来，以便于推送消息时可以找到消费者。
信道管理模块：一个TCP连接可能会有多个信道，当某个客户端想要关闭通信，关闭的不是连接，而是自己的信道，关闭信道的同时，我们需要将该客户端的订阅信息给删除。
连接管理模块：我们的服务器可能会收到多个客户端的连接，我们需要将这些连接保存起来。
Broker Server模块（搭建服务器）：对以上模块的整合，向客户端提供服务。

2，服务端模块汇总图

首先介绍服务器模块，当用户创建交换机和队列，以及绑定交换机和队列时，这些信息应该被持久化存储，这里使用的是sqlite数据库进行存储，以便于主机重启/程序重启后，这些信息可以恢复。同样，消息也需要持久化存储，因为一条消息被放入到消息队列后，可能还没有被消费，服务器就可能因为某些原因停止运行了，为了使服务器重启后，消息还存在，这里使用文件进行存储。

当将一条消息发布给消息队列时，这个消息首先是交给交换机，再根据交换机的类型以及路由匹配，将消息推送给绑定的队列，然后这条消息就会被保存下来，内存中保存一份，文件中也会保存。当这个消息推送给某个消费者客户端，并且收到确认响应后，才会从内存和文件中将这条消息删除掉。

3，客户端模块设计

消费者模块：一个客户端如果订阅了消息，那么他就是消费客户端，在向服务器发送订阅消息请求的时候，要包含本身的消费者信息。
信道管理模块：同样我们在发布消息/获取消息的时候，是通过信道来完成的。
连接管理模块：该模块提供各种API接口，比如创建交换机，创建队列等等所有的接口。

最后基于以上三个模块：实现发布客户端和订阅客户端。

4，客户端模块汇总图

七，工具类的实现

在本项目中，会使用到一些工具类，比如线程池，对数据库的操作，随机ID的生成等等。

所以，在这里先实现这些工具类，之后在项目中直接使用。

1，线程池的实现

1，前置知识

使用C++11中的异步操作，实现一个线程池。

std::future是C++11标准库中的一个模板类，它表示一个异步操作的执行结果。当我们在多线程编程中，使用异步操作来执行任务时，std::future可以帮助我们获取任务的执行结果。std::future会阻塞当前线程，直到异步任务执行完。

std::future提供了一种安全的方式来获取异步任务的执行结果，我们可以通过std::future::get()函数来获取任务的结果，此函数会阻塞当前线程，直到异步任务执行完。

std::future通常要与其他组件（如async/promise/packaged_task）搭配使用。 1，std::async是一个函数模板作用：异步启动任务，该函数会返回一个future对象，用来获取执行结果。异步任务的执行方式： std::launch::async：强制在新线程执行。 std::launch::deferred：延迟执行（调用get()时执行）。 2，std::promise是一个类模板调用这个类的成员函数get_future()，返回一个std::future对象。 3，std::packaged_task是一个类模板同样也可以调用成员函数get_future()，返回一个std::future对象。可以对可调用对象进行二次封装，将其结果绑定到std::future中

std::async与std::future搭配使用示例：

//使用future来获取异步线程的执行结果
#include <iostream>
#include <future>
#include <chrono>

int Add(int num1,int num2)
{
    std::this_thread::sleep_for(std::chrono::seconds(3));
    std::cout<<"开始执行！"<<std::endl;
    int resul=num1+num2;
    return resul;
}
int main()
{
    //std::launch::deferred  当调用get获取执行结果时，异步线程才会执行任务
    //std::launch::async 异步线程会立即执行任务
    // std::future<int> ret=std::async(std::launch::deferred,Add,11,22);
    std::future<int> ret=std::async(std::launch::async,Add,11,22);
    std::this_thread::sleep_for(std::chrono::seconds(1));
    int sum=ret.get();//获取执行结果
    std::cout<<sum<<std::endl;
    return 0;
}

std::promise与std::future搭配使用示例：

//promise模板类的使用
#include <iostream>
#include <future>
#include <thread>
void Add(int num1,int num2,std::promise<int>& pro)
{
    std::this_thread::sleep_for(std::chrono::seconds(3));
    int result=num1+num2;
    pro.set_value(result);
}
//通过promise对象使两个线程之间的结果得到同步
int main()
{
    std::promise<int> pro;

    std::future<int> fut=pro.get_future();

    std::thread thr(Add,11,22,std::ref(pro));
    int ret=fut.get();//会在这里阻塞住，直到pro中结果被设置了

    std::cout<<"sum:"<<ret<<std::endl;
    thr.join();

    return 0;
}

std::packaged_task与std::future搭配使用案例：

// packaged_task的使用
#include <iostream>
#include <future>
#include <memory>

int Add(int num1, int num2)
{
    std::this_thread::sleep_for(std::chrono::seconds(3));
    std::cout << "开始执行！" << std::endl;
    int resul = num1 + num2;
    return resul;
}
int main()
{
    // 可以通过packaged_task对象交给智能指针来管理，也就是封装成一个指针
    // 然后将该指针传入线程中，再解引用执行
    // 但是此时的packaged_task对象就需要从堆上new一个，而不是在栈上开辟
    // 因为线程的执行顺序不确定，可能packaged_task对象销毁了，才会执行线程函数
    //  std::shared_ptr<std::packaged_task<int(int, int)>> ptask(new std::packaged_task<int(int, int)>(Add));
    auto ptask = std::make_shared<std::packaged_task<int(int, int)>>(Add);
    std::thread thr([ptask]()
                    {
    std::this_thread::sleep_for(std::chrono::seconds(3));

        (*ptask)(11,22); });
    std::future<int> fut = ptask->get_future();
    int ret = fut.get(); // 同样会在这里阻塞
    std::cout << ret << std::endl;
    thr.join();

    // ————————————————————————————————————————————————————————————————————————————————————
    //  定义一个packaged_task对象，它可以用来管理一个可调用对象
    //   std::packaged_task<int(int,int)> task(Add);
    //   std::future<int> fut=task.get_future();
    //  //可以把该对象当作使一个可调用对象来执行
    //  task(11,22);
    //  int ret=fut.get();
    //  std::cout<<ret<<std::endl;

    // ——————————————————————————————————————————————————————————————————————————————————
    //  不能将task对象传入async，让异步线程去执行
    //   std::packaged_task<int(int,int)> task(Add);
    //   std::future<int> fut=task.get_future();
    //   std::async(std::launch::async,task,11,22);
    //   std::cout<<fut.get()<<std::endl;

    // ——————————————————————————————————————————————————————————————————————————————————
    // 也不能将task对象作为线程的入口函数
    //  std::packaged_task<int(int,int)> task(Add);
    //  std::future<int> fut=task.get_future();
    //  std::thread thr(task,11,22);
    //  std::cout<<fut.get()<<std::endl;
    //  thr.join();
    return 0;
}

2，线程池的实现

由于使用std::promise，需要将std::promise对象设置到任务的执行参数中，还需要将运行结果设置到该对象中，也就是我们要改变用户传入的函数，实现难度较高，不采用这种方式。而std::async，使用的是自己内部的工作线程，那我们创建的线程用来干什么呢？

所以，最后决定使用std::packaged_task与std::future来实现线程池。

线程池的实现思想：

用户传入要执行的任务（函数），以及要处理的数据（函数的参数），由线程池中的工作线程来完成异步执行。

要提供的操作：

push操作：用户向线程池中放入一个任务。
stop操作：停止线程池的运行。

要管理的成员：

任务池：保存用户传入的任务
工作线程池：保存一定数量的线程
互斥锁&条件变量：实现同步互斥
结束运行标志：以便于控制线程池的结束

// 线程池
#ifndef __M_THRPOOL_H__
#define __M_THRPOOL_H__

#include <iostream>
#include <vector>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <thread>
#include <functional>
#include <future>
#include <memory>

namespace xgmq
{
    class ThreadPool
    {
    public:
        using Funct = std::function<void()>;
        using ptr=std::shared_ptr<ThreadPool>;
        ThreadPool(const int thr_count = 3)
            : _stop(false)
        {
            for (int i = 0; i < thr_count; i++)
            {
                _thr_pool.emplace_back(&entry, this);
            }
        }
        // 向线程池中加一个任务
        template <typename F, typename... Args>
        auto push(F &&func, Args &&...args) -> std::future<decltype(func(args...))>
        {
            // 获取该函数的返回类型
            auto return_type = decltype(func(args...));
            // 将该函数的参数进行绑定
            auto func_t = std::bind(std::forward(func), std::forward(args)...);
            auto task = std::make_shared < std::packaged_task<return_type()>(func_t);
            std::future<return_type> fut = task->get_future();
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _task_pool.emplace_back([task]()
                                        { (*task)(); });
                _cond.notify_one();
            }
            return fut;
        }
        // 停止线程池的运行
        void stop()
        {
            if (_stop == true)
            {
                return;
            }
            _stop = true;
            _cond.notify_all(); // 唤醒所有进程
            for (int i = 0; i < _thr_pool.size(); i++)
            {
                _thr_pool[i].join();
            }
        }
        ~ThreadPool()
        {
            stop();
        }

    private:
        // 线程入口函数
        void entry()
        {
            while (!_stop)
            {
                std::vector<Funct> _tmp_pool;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _cond.wait(lock, [this]()
                               { return !_stop || _task_pool.size() > 0; });
                    _tmp_pool.swap(_task_pool);
                }
                for (auto &task : _tmp_pool)
                {
                    task();
                }
            }
        }

    private:
        std::atomic<bool> _stop; // 线程池是否停止运行
        std::mutex _mutex;
        std::condition_variable _cond;
        std::vector<Funct> _task_pool;      // 任务池
        std::vector<std::thread> _thr_pool; // 线程池
    };
}
#endif

2，日志打印

在项目中，会涉及到一些地方的日志信息打印，来观察程序的运行状态。

在这里，要求进行日志打印的时候，要带上行号和文件名。同时日志打印也有等级的划分，有调试等级的日志打印，有警告等级的日志打印，也有错误等级的日志打印。

用到的关键技术：C语言库中的strftime函数。 size_t strftime(char *str, size_t maxsize, const char *format, const struct tm *timeptr) 这个函数可以将传入的时间，按照指定格式，组织成字符串。 str标识最终形成的字符串，maxsize表示str的最大字符个数，format表示生成的字符串的格式，最后一个参数是一个tm类型的结构体，这个结构体内部就包含了时间，比如年月日，时分秒，微秒......

使用示例：

//获取当前时间
time_t t=time(nullptr);
//使用当前时间构造一个struct tm类型的结构体
struct tm* ptm=localtime(&t);

char str[32];
strftime(str,31,"%H:%M:%S",ptm);//时间格式为 时:分:秒

// 日志打印工具
#ifndef __M_LOGGER_H__
#define __M_LOGGER_H__
#include <iostream>
#include <cstdio>
#include <ctime>

#define DEBUG_LEVEL 0
#define INFO_LEVEL 1
#define ERR_LEVEL 2
#define DEFAULT_LEVEL DEUBG_LEVEL
#define LOG(leve_str, level, format, ...)                                                                   \
    {                                                                                                       \
        if (level >= DEFAULT_LEVEL)                                                                         \
        {                                                                                                   \
            time_t t = time(nullptr);                                                                       \
            struct tm *ptm = localtime(&t);                                                                 \
            char time_str[32];                                                                              \
            strftime(time_str, 31, "%H:%M:%S", ptm);                                                        \
            printf("[%s][%s][%s:%d]\t" format "\n", leve_str, time_str, __FILE__, __LINE__, ##__VA_ARGS__); \
        }                                                                                                   \
    }

#define DLOG(format,...) LOG("DEBUG",DEBUG_LEVEL,format,##__VA_ARGS__)
#define ILOG(format,...) LOG("INFO",INFO_LEVEL,format,##__VA_ARGS__)
#define ELOG(format,...) LOG("ERR",ERR_LEVEL,format,##__VA_ARGS__)

#endif

3，sqlite数据库操作的封装

    class SqliteHelper
    {
    public:
        typedef int (*callback)(void *, int, char **, char **);
        SqliteHelper(const std::string &dbfile)
            : _dbfile(dbfile), pb(nullptr)
        {
        }
        // 打开数据库文件
        // 传入打开时的安全等级,默认是串行化
        void open(int safe = SQLITE_OPEN_FULLMUTEX)
        {
            // 以可读可写方式，不存在则创建的方式打开
            int ret = sqlite3_open_v2(_dbfile.c_str(), &pb,
                                      SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | safe, nullptr);
            if (ret != SQLITE_OK)
            {
                DLOG("打开数据文件失败：%s", _dbfile.c_str());
                return;
            }
        }
        // 执行sql语句
        bool exec(const std::string &sql, callback cb, void *arg)
        {
            int ret = sqlite3_exec(pb, sql.c_str(), cb, arg, nullptr);
            if (ret != SQLITE_OK)
            {
                DLOG("执行语句失败：%s", sql.c_str());
                return false;
            }
            return true;
        }
        // 关闭数据库文件
        void close()
        {
            if (pb != nullptr)
            {
                sqlite3_close_v2(pb);
            }
        }

    private:
        std::string _dbfile;
        sqlite3 *pb;
    };

4，字符串分割方法的实现

前面提到过，消息是先交给交换机，再根据交换机的类型和路由匹配，决定将这条消息该交给哪个队列。

如果交换机的类型是主题交换模式（Topic），那么就要进行路由匹配（如何匹配在后面模块处详解），只有当binding_key与routing_key匹配成功时，才将该消息交给这个队列。

在这部分会涉及到将一个字符串按照指定字符进行分割的操作，所以在这里先实现。

    class StrSplit
    {
        public:
        static size_t split(const std::string& str,const std::string& sep,std::vector<std::string>& result)
        {   
            size_t pos=0,idx=0;
            while(idx<str.size())
            {
                pos=str.find(sep);
                if(pos==std::string::npos)
                {
                    result.push_back(str.substr(idx));
                    return result.size();
                }
                if(idx==pos)
                {
                    idx=pos+sep.size();
                    continue;
                }
                std::string tmp=str.substr(idx,pos-idx);
                result.push_back(tmp);
                idx=pos+sep.size();
            }
            return result.size();
        }
    };

5，uuid生成器

UUID，也叫通用唯一识别码，通常由32位16进制数字字符组成。在进行消息的发布和消息的确认到的时候，每个消息是有一个唯一ID的，来标识消息的唯一性。在这里不使用一个自增的整数来表示，是因为如果消息数量太多的话，这个数字是有可能会越界的。所以，我们决定使用一个字符串来表示这个ID，那么现在我们就需要保证整个字符串不能出现重复。 uuid的标准形式为：8-4-4-4-12的32个字符，如550e8400-e29b-41d4-a716-446655440000.

实现思路：首先生成8个随机数字，范围在[0,255]之间，因为1个字节可以表示的数据范围就是[0,255]，这样就可以生成8字节的数字，再加上8字节的序号，构成16字节的数字，然后将这16字节的数字转化成32位16进制字符。

1，std::random_device这是c++11为我们提供生成随机数的方式，通过这个类定义对象生成的随机数是一种机器随机数，也就是通过硬件来生成的，效率较低。 2，标准库中还提供了一种生成随机数的算法：std::mt19937_64，通过这个类生成的随机数范围是在[0,2^19937]之间，不过他生成的随机数是一个伪随机数。 3，所以，我们可以将这两者结合，先使用std::random_device生成一个随机数，将这个随机数设置为std::mt19937_64的随机数种子。之后我们使用std::mt19937_64生成随机数即可。 4，此时生成的随机数很大，而我们要的是[0,255]之间的。我们可以使用标准库中的std::uniform_int_distribution来限定生成随机数的数字区间。

    class UUID
    {
    public:
        static std::string uuid()
        {
            // 先生成随机数种子
            std::random_device rd;
            std::mt19937_64 genorator(rd()); // 生成伪随机数
            // 生成0-255之间的数字
            std::uniform_int_distribution<int> distribution(0, 255);
            std::stringstream ss;
            for (int i = 0; i < 8; i++)
            {
                ss << std::setw(2) << std::setfill('0') << std::hex << distribution(genorator);
                if (i == 3 || i == 5 || i == 7)
                {
                    ss << "-";
                }
            }

            // 生成序号
            static std::atomic<int> seq(1);
            size_t num = seq.fetch_add(1);

            // 将序号添加到字符串后
            // 将序号的每一字节转化为16进制字符
            // size_t 在64位下是8字节
            for (int i = 7; i >= 0; i--)
            {
                ss << std::setw(2) << std::setfill('0') << std::hex << ((num >> i * 8) & 0xff);
                if (i == 6)
                {
                    ss << "-";
                }
            }
            return ss.str();
        }
    };

6，文件操作类的实现

前面服务器模块设计中提到过，我们使用文件来保存消息，在这里肯定就会涉及到很多文件相关的操作。

需要封装的操作：

判断文件是否存在
获取文件大小
文件读/写
文件的创建/删除
目录的创建/删除

使用C语言库中提供的函数int stat(const char* path,struct stat* buf)，可以获取一个文件的信息，比如文件的大小，文件的创建时间，文件的inode编号等等。返回值等于0就表示函数执行成功。

文件的信息就存放于struct stat结构体中。

    class FileHelper
    {
        public:
            FileHelper(const std::string& filename)
            :_filename(filename)
            {}
            //判断问价是否存在
            bool exists()
            {
                struct stat st;
                return (stat(_filename.c_str(),&st)==0);
            }
            //获取文件大小
            size_t size()
            {
                struct stat st;
                int n=stat(_filename.c_str(),&st);
                if(n!=0)
                {
                    DLOG("获取文件信息失败：%s",_filename.c_str());
                    return 0;
                }
                return st.st_size;
            }
            //文件的读取
            //从指定位置开始，读取指定长度的数据
            bool read(char* buff,size_t offset,size_t len)
            {
                std::ifstream ifs(_filename.c_str(),std::ios::binary|std::ios::in);
                if(ifs.is_open()==false)
                {
                    ELOG("打开文件失败:%s",_filename.c_str());
                    return false;
                }
                //移动到指定位置
                ifs.seekg(offset,std::ios::beg);
                //开始读取
                ifs.read(buff,len);
                if(ifs.good()==false)
                {
                    ELOG("读取文件失败:%s",_filename.c_str());
                    return false;
                }
                return true;
            }
            //文件的读取，读取全部数据
            bool read(std::string& buff)
            {
                buff.resize(size());
                return read(&buff[0],0,size());
            }
            //向文件指定位置写入指定长度的数据
            bool write(const char* data,size_t offset,size_t len)
            {
                std::fstream fs(_filename,std::ios::binary|std::ios::in|std::ios::out);
                if(fs.is_open()==false)
                {
                    ELOG("打开文件失败：%s",_filename.c_str());
                    return false;
                }
                //移动到指定位置
                fs.seekg(offset,std::ios::beg);
                fs.write(data,len);
                if(fs.good()==false)
                {
                    ELOG("写入文件失败:%s",_filename.c_str());
                    return false;
                }
                return true;
            }
            //向文件开始处写入数据
            bool write(const std::string& data)
            {
                return write(data.c_str(),0,size());
            }

            //创建文件
            static bool createFile(const std::string& filename)
            {
                std::ifstream ifs(filename,std::ios::binary);
                if(ifs.is_open()==false)
                {
                    ELOG("创建文件失败:%s",filename.c_str());
                    return false;
                }
                ifs.close();
                return true;
            }
            //删除文件
            static void removeFile(const std::string& filename)
            {
                ::remove(filename.c_str());
            }
            //获取文件的父路径
            static std::string parentDirectory(const std::string& path)
            {
                //aaa/bbb/ccc/a.txt
                size_t pos=path.find_last_of("/");
                if(pos==std::string::npos)
                {
                    return "./";
                }
                std::string parent=path.substr(0,pos);
                return parent;
            }
            //创建路径
            static bool createDirectory(const std::string& path)
            {
                //aaa/bbb/ccc 可能存在多级目录 
                size_t pos=0,idx=0;
                while(idx<path.size())
                {
                    pos=path.find("/");
                    if(pos==std::string::npos)
                    {
                        return mkdir(path.c_str(),0775)==0;
                    }
                    std::string parent=path.substr(0,pos);
                    int ret=mkdir(parent.c_str(),0775);
                    //目录如果已经存在，在创建的时候返回置为非0,并且错误码被设置
                    if(ret!=0&&errno!=EEXIST)
                    {
                        ELOG("创建目录失败:%s",parent.c_str());
                        return  false;
                    }
                    idx=pos+1;
                }
                return true;
            }
            //删除路径
            static bool removeDirectory(const std::string& path)
            {
                std::string cmd="rm -rf "+path;
                return ::system(cmd.c_str());
            }
            //修改文件名称
            bool rename(const std::string& nname)
            {
                return (::rename(_filename.c_str(),nname.c_str())==0);
            }
        private:
        std::string _filename;
    };

八，项目实现

项目实现部分不展示代码，代码仓库附在最后。

1，服务器模块

1，交换机模块

首先是交换机数据结构的定义：

对于一个交换机，它一个具有以下属性：

交换机的名称（交换机唯一标识）
交换机的类型（主题交换OR直接交换OR广播交换）
交换机的存储方式（是否需要将交换机的信息进行持久化存储）

接着是交换机信息持久化类的实现：使用sqlite数据库来保存

最后是对交换机管理类的实现：向外提供声明交换机，移除交换机，获取指定交换机的方法。

2，队列模块

该模块与交换机模块类似：

队列结构的定义：

队列的名称（唯一标识）
存储方式（是否持久化）
是否独占标志（只能有一个客户端来订阅该队列）
是否自动删除标志（当创建该队列的客户端退出后，是否自动删除该队列）

还有队列信息持久化类的实现：使用sqlite数据库来存储。

最后是队列管理类的实现：向外提供声明队列，移除队列，获取指定队列的方法。

3，绑定关系

描述交换机与队列之间的关系，这个模块的设计与前面两个相似。

首先是绑定关系结构的定义：

应该包含的成员：交换机名称，队列名称，还有binding_key（交换机进行消息路由的时候会使用到）

还有绑定关系的持久化类，交换机和哪些队列绑定了也要持久化存储，这样在服务器重启，仍然可以恢复。

最后是一个管理类，向外提供建立一组绑定关系，解除一组绑定关系，解除指定交换机的所有绑定关系，解除指定队列的所有绑定关系，获取一组绑定关系，获取指定交换机的所有绑定关系等操作。

4，消息模块

消息在进行持久化存储的时候，使用的是文件存储，这就会涉及到序列化和反序列工作，所以这里使用protobuf来描述消息的结构，生成消息的结构定义以及序列化和反序列化功能的实现。

编写的message.proto文件内容（在这里顺便也定义了交换机的类型）：

一个完整的消息包含有效载荷和其他参数

消息的有效载荷包括：这个消息的属性，消息的正文，以及这个消息是否有效。
消息的属性包括消息的id，消息的routing_key，以及消息的投递模式（是否持久化存储）。
其他参数：这个消息在文件中存储的位置（偏移量）和这个消息的长度。

syntax="proto3";

package xgmq;
//交换机类型
enum ExchangeType
{
    UNKNOWN_TYPE=0;
    DIRECT=1;
    TOPIC=2;
    FANOUT=3;
};
enum DeliveryMode
{
    UNKNOWN=0;
    UNDURABLE=1;
    DURABLE=2;
};
message basicProperties
{
    string id=1;
    DeliveryMode 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;
};

消息的持久化实现：

在这里，使用文件来存储消息，但是这里如果使用一个文件来存储所有的消息，那么在访问的时候，就需要加锁，锁竞争会非常频繁，所以不适用这种方式。消息的载体是队列，消息最终都是要交给队列的，所以在消息进行持久化时，以队列为单位，每个队列一个文件，来存储该队列上的消息，这样当访问不同队列的消息时，就不存锁竞争问题时，访问同一个队列的消息时，还是需要加锁的，这样就可以大大降低锁竞争的概率。

在实现消息的持久化存储，就必然会涉及到消息的读和写，那么此时就会涉及到怎么写，怎么读的问题，也就是协议。这里设定，在写文件的时候，按照格式：【8个字节的消息长度消息内容】这样的方式进行写入。那么在读取的时候，就先读前8个字节，知道消息的长度之后，就可以读取后面消息的内容。这样做就可以解决类似Tcp粘包问题。

新增一个消息，我们直接在文件后面追加写这个消息即可，但如果时删除消息，此时我们的做法是，找到这条消息在文件中的位置（偏移量offset），将这条消息的是否有效标志为设为无效，然后写入文件，覆盖掉之前的数据，这样就算一次删除消息的操作。

但是删除完一条消息之后，文件的大小没变，如果此时文件中无效消息数量超过总消息数量的一半，那么此时就需要进行垃圾回收，此时我们创建一个临时文件，然后读取源文件，将有效消息写入新文件，最后在讲源文件删除掉，临时文件重命名为源文件名，这样一个删除操作才算完成。

5，虚拟机模块

虚拟机模块是对以上四个模块的整合，使用上述四个模块的管理句柄，向外提供服务：

声明/销毁交换机，声明/删除队列，绑定/解绑，发布一条消息，订阅一条消息，确认一条消息，还有获取所有队列，获取指定交换机的所有绑定信息等操作。

6，路由匹配模块

当一条消息被发布到交换机上之后，需要判断交换机的类型，如果是主题交换（Topic），需要进行路由匹配，决定这条消息应该交给哪些队列。

这个模块只用来提供功能：

判断routing_key是否合法。
判断binding_key是否合法。
判断binding_key与routing_key是否匹配成功（路由匹配功能）。

routing_key和binding_key都是以'.'作为分割的字符串。

对于routing_key来说：合法字符包含（a~z,A~Z,0~9, '.' , '_'）等字符。

对于binding_key来说：合法字符包含（a~z,A~Z,0~9, '.' , '_'），除了这些外，还包含通配符'*'和'#'。但是这两个通配符不能连续出现，并且单独出现的，不能和其他字符一起出现。'*'可以匹配任意的一个字符，'#'可以匹配0个或多个任意字符。

路由匹配规则：

先将routing_key和binding_key按照'.'进行分割，使用数组保存分割后的字符串（单词）。也就是说现在有两个数组：rout_arr保存routing_key分割后得到的每个单词，bind_arr保存binding_key分割后的每个单词。

然后让rout_arr的每个单词与bind_arr的每个单词进行比对。(使用动态规划算法)

定义状态表示：dp[i][j]表示rout_arr的第i个单词与bind_arr的第j个单词是否匹配成功。0表示匹配失败，1表示匹配成功。

结果：dp表最后一个位置的值表示最终是否匹配成功。

状态转移方程的推导：

示例1：binding_key="bbb.ddd"，routing_key="aaa.ddd"，dp表结果如下：

	aaa	ddd
bbb	0	0
ddd	0	0

当binding_key与routing_key的第一个单词进行比较的时候，bbb与aaa匹配失败。当binding_key与routing_key的第二个单词进行比较的时候，本来是匹配成功的，但由于前一个单词匹配失败，所以这里也就失败了。所以可以得出一个状态转移：如果dp[i][j]是匹配成功的，那么还需要看上一个单词的匹配结果，dp[i][j]=dp[i-1][j-1]

示例2：binding_key="#"，routing_key="aaa.bbb"，下面的dp表多开一行一列，可以避免越界，但是dp[0][0]要置为 1，否则就会影响后面结果判断的正确性。

		aaa	bbb
	1	0	0
#	0	1	0->1

#与aaa匹配成功，dp[1][1]=1，同时继承与dp[i-1][j-1]=1，所以dp[1][1]=1。

#与bbb匹配成功，dp[2][2]=1，但由于dp[i-1][j-1]=0，所以dp[2][2]=0。

但是按理来说，#是匹配任意多个字符，是可以匹配成功 aaa.bbb的，所以这里还需要从当前位置的左边继承结果。可以得出结论，如果是通配符"#"，dp[i][j]不仅是要从dp[i-1][j-1]继承结果，还要从dp[i][j-1]位置继承结果。

所以，如果出现"#"，dp[i][j]=dp[i-1][j-1] | dp[i][j-1]。

示例3：binding_key="aaa.#"，routing_key="aaa"

		aaa
	1	0
aaa	0	1
#	0	0->1

同理，上面的示例可以看出，当出现"#"时，还可以从上方继承结果。

所以dp[i][j]=dp[i-1][j-1] | dp[i][j-1] | dp[i-1][j]。

"#"还有一个特殊之处，就是如果binding_key是以"#"开头，那么dp表第一列需置为1，否则会影响后续填表的正确性。

总结：

当两个单词匹配成功时，从左上方继承结果：dp[i-1][j-1]。
当遇到通配符"#"时：dp[i][j]=dp[i-1][j-1] | dp[i][j-1] | dp[i-1][j]。
当binding_key以 "#"开始时：需要将对应行的第一列位置初始化为1.
当遇到通配符"*"时，"*"是用来匹配任意一个单词的，所以可以当作第一种情况。

7，消费者管理模块

当有消费者订阅了服务器上的某个队列时，就需要将该消费者的信息保存起来，当这个队列上有了消息之后，就会把消息推送给这个队列，因此需要将消费者的信息记录下来。

注意：一个队列可能会被多个消费者订阅，当队列中有消息时，会采用RR轮转的方式，选择一个消费者，然后将消息推送给消费者，从而实现负载均衡。

在前面提到过，客户端与服务器在及逆行交互的时候，使用的是信道，信道给我们提供服务的。当客户端想要订阅一个队列的消息时，是通过信道发送对应的请求，然后在服务器这边，就会创建一个消费者对象，然后将这个消费者队列"绑定"在一起，当队列中有消息的时候，就找到该消费者，然后找到对应的信道，将响应发送给客户端。

一个消费者属于某个信道的，当信道关闭或者连接关闭时，都要删除这个消费者。

定义消费者结构：

消费者名称：消费者唯一标识
队列名称：该消费者所订阅的队列
是否自动应答标志：该消费者获取到消息后，是否需要收到对这个消息进行确认。
一个回调函数：当该队列有消息后，先找对应的消费者，再调用消费者内部的回调函数，将这条消息发送给客户端。

8，信道管理模块

信道是针对连接更细粒度的一个通信通道，多个信道可以使用同一个连接来进行通信，但是同一个连接的信道之间是相互独立的。

信道建立在TCP连接之上的，是抽象出来的概念，用户通过信道向服务器发送请求，同时服务器也通过信道向客户端发送响应。

信道应该提供以下的服务：

创建/删除交换机
创建/删除队列
绑定/解绑
发布消息
订阅消息
对消息进行确认
取消订阅

信道结构的定义：

信道ID：信道的唯一标识
信道关联的消费者：信道提供了订阅消息的服务，当某个用户订阅了队列的消息，此时就会产生一个消费者。当队列中有消息的时候，就会把消息推送给消费者。注意：如果信道关闭了，就需要将这条信道上的消费者也删除掉，避免造成内存泄漏的问题。
信道关联的连接：毕竟真正进行通信的时候，还是要使用到底层TCP连接的。
虚拟机句柄：完成上述各种操作，创建/删除交换机......
线程池句柄：当一个消息被推送到某个队列上后，需要将这条消息推送给订阅的消费者，这个各种交给线程池来完成。
消费者管理句柄：用于删除消费者信息。信道关闭/取消订阅的时候，要删除消费者信息。（防止内存泄漏）
protobuf协议处理句柄：网络通信中的协议处理。

接下来还需要对信道进行管理：

用哈希表来管理信道——信道ID和信道对象的映射。
同时提供打开信道，关闭信道，获取信道的功能。

9，应用层协议设计

应用层协议格式如下：

len：4个字节，表示整个报文的长度。
namelen：表示typeName的长度。
typeName：表示请求/响应报文的类型名称。（比如创建交换机请求的类型，创建队列请求的类型）
protobufData：表示请求/响应数据通过protobuf序列化之后的二进制数据。
checkSum：校验和。

上述的应用层协议，其实就是muduo库中，陈硕大佬定义的一种应用层协议。我们使用muduo库搭建服务器，所以也就使用这个应用层协议来实现网络通信。

接下来需要定义请求/响应参数（格式）：

syntax="proto3";

package xgmq;

//需要用到消息的属性，所以引入了消息模块
import "message.proto";

//信道的打开与关闭请求
message openChannelRequest
{
    string rid=1;//请求id
    string cid=2;//信道id
};

message closeChannelRequest
{
    string rid=1;
    string cid=2;
};

//交换机的创建与销毁请求
message declareExchangeRequest
{
    string rid=1;
    string cid=2;
    string exchange_name=3;//交换机名称
    ExchangeType exchange_type=4;//交换机类型
    bool durable=5;//是否持久化
};

message deleteExchangeRequest
{
    string rid=1;
    string cid=2;
    string exchange_name=3;
};

//队列的创建与销毁请求
message declareQueueRequest
{
    string rid=1;
    string cid=2;
    string queue_name=3;
    bool exclusive=4;
    bool durable=5;
    bool auto_delete=6;
};

message deleteQueueRequest
{
    string rid=1;
    string cid=2;
    string queue_name=3;
};

//队列的绑定与解绑请求
message queueBindRequest
{
    string rid=1;
    string cid=2;
    string exchange_name=3;
    string queue_name=4;
    string binding_key=5;
};
message queueUnBindRequest
{
    string rid=1;
    string cid=2;
    string exchange_name=3;
    string queue_name=4;
};

//消息的发布
message basicPublishRequest
{
    string rid=1;
    string cid=2;
    string exchange_name=3;
    string body=4;
    basicProperties perties=5;
};

//消息的确认请求
message basicAckRequest
{
    string rid=1;
    string cid=2;
    string msg_id=3;
    string queue_name=4;
};

//队列订阅请求
message basicConsumeRequest
{
    string rid=1;
    string cid=2;
    string consume_tag=3;//消费者名称
    bool auto_ack=4;//是否自动应答
    string queue_name=5;//队列名
};

//队列的取消订阅
message basicCancelRequest
{
    string rid=1;
    string cid=2;
    string queue_name=3;
    string consum_tag=4;
};

//消息的推送响应
message basicConsumeResponse
{
    string cid=1;
    string consum_tag=2;//消费者名称
    string body=3;//消息内容
    basicProperties properties=4;//消息的属性
};

//通用响应
message basicCommonResponse
{
    string rid=1;
    string cid=2;
    bool ok=3;
};

10，连接管理模块

服务器可能会收到很多客户端的连接，所以先将这些连接管理起来，每个TCP连接都对应有一个信道管理句柄，这样就实现了一个TCP连接，可以创建多个信道。

而muduo中的连接connection对象，是没有这个功能的，使用这个connection对象只是来完成网络通信的。

所以，这里我们需要再将muduo库中的connection连接封装一次，提供创建信道/关闭信道的功能。

提供的操作：

创建/关闭信道。

成员信息：

我们通过连接创建信道，一个连接的成员就是创建信道时所需的参数，比如connection连接，虚拟机句柄等等。

信道管理句柄（实现信道的增删查）：用来打开/关闭/获取信道。当某个连接断开时，相关的信道信息也应该被释放掉，避免内存泄漏。（信道释放了，那么这条信道上的消费者也应该被释放掉，避免内存泄漏，不过这个工作已经在信道模块处理过了，当一个信道对象析构时，其内部会通过消费者管理句柄删除该消费者对象）
连接关联的实际用于通信的 muduo::net::Connection 连接
protobuf 协议处理的句柄
消费者管理句柄
虚拟机句柄
异步工作线程池句柄

连接管理的实现：

使用哈希表来管理连接，使用muduo::net::TcpConnectionPtr作为key，而我们封装的连接对象作为value。

当一条Tcp连接建立成功之后，我们需要为这个Tcp连接建立一个我们封装的连接对象，通过这个对象，进行打开信道，关闭信道，获取信道的操作，这样通过一个Tcp连接，就可以建立多个信道了。

11，brokerserver模块

该模块使用muduo库进行服务器的搭建，只需设置请求的处理回调函数即可。

服务器一旦启动，需要完成对消费者管理结构的初始化，因为当由客户端订阅队列消息时，需要将消费者信息添加到消费者管理结构中，如果不初始化，则会出现对空对象访问的错误。

2，客户端模块

1，订阅者模块（消费者模块）

这个模块并不直接向用户展示，这个模块只起到一个角色描述的作用，表示这是一个消费者。如果是生产者发布消息，那它就不需要这个。

2，信道模块

在客户端这边同时也是有信道的，用户通过信道向服务器发送请求，所以，信道同样也要提供服务器模块的相关服务：创建/删除交换机，创建/删除队列......，不过这些服务的内部，只是构建出一个请求，然后发送给服务器，服务器收到后，进行解析处理，所以真正做事的是服务器。

信道的管理：向用户提供创建信道，关闭信道，获取信道的操作。当用户调用创建信道的操作时，同时其内部会向服务器发送一个创建信道的请求，至此服务器和客户端都存在信道了，而客户端这边的信道是为用户提供服务的，而服务器这边的信道是为客户端提供服务的。

注意：当向服务器发送一个请求后，需要等待这个请求被服务器处理后，并接收到服务器发送来的响应，才能进行下一步操作。由于muduo库中的操作都是异步执行的，当要执行如下操作时：创建交换机1，创建队列1，绑定交换机1和队列1，结果可能是：创建队列1和绑定请求先发送到了服务器，交换机1还没有创建，此时就会出现绑定失败的问题。所以，我们需要等待创建交换机1的请求接收到对应的响应之后，再发送下一个请求。

这里使用哈希表来存储响应id和响应的映射。