项目介绍

介绍

本项目已上传，后期会做扩展：Gitee获取完整项目源码

该项目是仿照RabitMQ实现简版的消息队列。主要是解决了普通生产消费者模型只能在单主机上生产消费模型的缺点。该项目是可以进行跨网络传输生产与消费，实现不同主机间的数据交换。

RabbitMQ具备生产者消费者模型的特点

• 强解耦
• 忙闲不均
• 高并发

同时又具备了

• 跨主机间通信
• 自动推送消息
• 传输的可靠性，确认应答
• 根据匹配规则，将消息推送给不同消费者
• 垃圾回收

开发环境：

• Linux / vsCode / g++ / gdb

技术选型

• 开发语言C++
• 序列化框架Protobuf
• 网络通信：自定义应用层+muduo库 ：封装tcp长连接的封装+epoll事件驱动，并发服务器
• 数据存储：文件   、SQLite3
• 单元测试框架：GTest

项目演示

生产者推送消息

服务启动。（左上）

创建消费者1，消费者客户端声明交换机、队列以及绑定的关系，订阅指定的队列queue1（右上）

创建消费者2，消费者客户端声明交换机、队列以及绑定的关系，订阅指定的队列queue2（右下）

创建生产者客户端，往交换机中推送消息，分别推送不同routingkey的消息。

服务器接收到客户端的请求，在服务端声明交换机、队列、绑定交换机和队列、创建消费者、信道、连接管理等。将接收到的消息根据路由匹配规则推送到某一个队列上，异步线程池调用任务取出任务和消费者进行推送消息。

交换机

交换机描述

交换机主要是用来接收消息，根据消息的类型，选择对某些队列转发。要实现消息的转发，必须记录交换机的属性，比如是否持久化，自动刷删除等，如果是持久化，那么在下一次打开交换机的时候，交换机必须要恢复数据。所有要先对交换机描述。另一方面交换机不止有一台，也就意味着在内存中存在多个交换机，需要我们组织管理起来。

交换机的成员变量

• 交换机的名称（主要是方便查找，标识）
• 推送类型（直接，广播，主题）
• 是否持久化
• 是否自动删除 
• 其它删除 key=val ,key=val ......

交换机提供设置其它参数的对外，将来想给交换机设置参数，只需要传入key=val&key=val的键值对字符串，就能完成就交换机的其它参数的插入。

同时也能够将其它参数通过接口，以字符串拼接的形式得到，主要是为了在数据库中能够读写。

交换机数据在硬盘中存储。（主要为了恢复）

 如果交换机每次重启，之前的连接就全部断开，消息全部丢失，需要重新建立连接，达成共识，形成完整的交换机。那么成本就太高了，为了能够高效的运转，将交换机的数据保存在硬件中，形成断电不丢失。

如果交换机被重启，交换机就会先到磁盘中，读取是否有需要被加载到内存中的机器。同时，对于一条消息来说，如果设置了持久化，就必须在磁盘中也保存一份。

对于交换机在磁盘中的管理，选择的是数据库中的SQLite3，轻量级，便捷性的本地程序。

交换机的持久化

接下来将设置接口，对交换机在SQLite中管理。

SQLite主要接口是open(创建库)，close(关闭库)，exec(执行sql语句)。在磁盘管理类中，必然要保存数据库的句柄，为的就是利用数据库对交换机进行增删查改。

成员

• sqlite句柄

对外接口

• 创建/删除数据库
• sqlite新增交换机
• sqlite删除交换机
• 交换机复原

介绍一下交换机恢复接口

我们希望将交换机保存在map中，以map<name,machine>的形式将交换机返回，所以sqlite执行语句的时候，我们传入一个回调函数和map的指针。

回调函数中，主要通过构建share_ptr的智能指针机器对象，将查询到的结果，通过row[i]，设置到机器的每一个属性上，最后将map返回。

---

交换机的管理 ：实际对外接口

上文介绍的交换机的属性，交换机在磁盘中的存储，都是对内的接口，我们是不暴露出去的。真正想对交换机的操作，实际上只能在内存中，我们提供对外声明交换机、查询交换机，删除交换机，就是最简单的增加、删除、查询。因为后续对交换机的访问呢，设计到多线程，可能存在线程安全，所以在增加、删除、查询中必然要加锁保护。

管理的成员：

• mutex;
• 磁盘管理类；
• map<name,machine>在内存中管理

对外的接口

• 声明交换机
• 删除交换机
• 获取指定交换机
• 交换机是否存在

这里的重点是在构造函数中，对已经持久化机器的恢复。如果重启交换器，所有内存数据必然为空，磁盘管理类中会读取数据库，保存交换机类的对象，返回到上层，具体来说就是返回到Map<name,machine>中管理。所以在交换机的恢复接口来说，它的返回值与内存中的实际管理对应上了，是一个很巧妙的设计。

对于增删查，都必须添加互斥锁保护！ 

 介绍一下声明交换机。

声明交换机，如果不存在就新增，如果已经存在交换机就退出，不做操作。

通过给定的交换机参数，在内存中查找。

构建智能指针对象，将数据插入到内存中，如果消息选择持久化，将将消息插入到sql中。

---

GTest测试

通过GTest，对单元模块测试。

分模块测试，先进行插入测试。表的打开正常，也能正常插入数据。 

对查询结果进行测试，在recovery中出现段错误，通过gdb+bt查看堆栈，将问题锁定。排查后发现在构建机器时，将智能指针make_shared写成std_ptr，导致非法访问。

最后进行删除，移除表的测试。

---

队列

队列数据描述

MQ消息队列中队列是一个中间件，用来接收交换机的消息。对已经广播的数据进行存储，如果有消费者订阅了指定的队列，那么就将消息进行推送。队列数据管理，本质就是一个先描述再组织的形式，先描述队列的信息，在把内存中或者磁盘中有哪些数据管理起来。

不难发现，队列的数据管理与交换机是基本一致的。队列的属性，必然包含队列的名称，是否独占一个队列（针对客户端来说，是否自己享有这个队列的所有数据）。为了提高消息队列的效率，以便重启后不丢失，需要选择是否持久化。另外，当一个队列所有的客户端都取消订阅后，希望队列能做到自动清除（删除数据库的表，清空在内存中的数据）.......和交换机一样，我们保留其它参数Args，处理特殊情况，便于扩展功能。

将队列想象成连接交换机和客户端通道的机器，会更好理解，它的接口设计与交换机基本是一致的，这里就简单赘述。

队列的属性

• 队列名称（唯一标志）
• 是否独占
• 是否持久化
• 自动删除
• 其它参数

其中，同样提供俩个接口分别是SetArgs()和GetArgs()。SetArgs是专门为获取指定队列服务的，select查询到数据库中的指定队列后，调用SetArgs()接口将字符串Key=Val格式的字符串格式化插入到机器中。GetArgs则是相反的操作，将键值对，形成字符串风格，方便后续的插入。

---

队列的持久化

——为了恢复队列

和交换机一样，将队列的管理即在内存中维护，也在磁盘中通过数据库维护。这么做的目的是，提高效率，不需要在每一次新增，配置队列。

接口的设计依旧是和交换机一致，需要数据库的句柄，才能对数据库的数据操作。接口包括，创建表，删除表，新增队列，删除队列，队列的恢复（获取所有的队列，保存在map内容中）。

成员：

• 数据库句柄

接口：

• 创建/删除表
• 新增/删除机器
• 队列的恢复recovery

这些接口大部分就是对数据库的简单操作，就快速浏览一下。 

---

 队列的管理：实际的对外接口

新增/删除队列实际上是在内存操作的，我们只需要调用管理类的接口，指定创建的格式，就能快速创建一个队列。另外管理中还包括了将磁盘中的队列恢复。

大致上队列的管理和交换机的管理是一样的。

成员：

• 互斥锁（多线程访问）
• mapper（磁盘管理）
• map（内存中管理）

对外接口：

• 声明/删除机器
• 是否存在
• 队列的总数
• 查询指定队列

---

绑定

绑定：连接交换机和队列

当生产者将消息创造出来，将消息交给交换机，交换机会根据消息的匹配方式（全广播，直接推送，主题推送）往某一队列里面发数据。而等到真正有消息的时候，再去建立 “信道” —交换机和队列的匹配，会严重影响传输的效率。所以就要提前建立交换机和队列的关联。

建立交换机和队列的关联，也称绑定。一个交换机可以将消息分发给多个队列，交换机和队列的关系是 1 ：n。同时又能直接通过交换机找到某一条队列的连接 （1 ：1）。因此要存储这种一对多的关系，是利用map<echange,queue_map>。 建立一个交换机的map，通过交换机map，找到指定交换机对应绑定的队列map。好处就是在智能指针管理下的插入，删除非常便捷。而不是再简单的一对 map<exchange,queue>。

绑定应该有的管理是交换机名称，队列名称，binding_key（广播的方式），是否持久化。关于绑定的持久化，是在交换机对应持久化，队列都持久化的前提下才有必要的。但是这一个判断我们不直接获取交换机和队列的属性，而是交给用户的判断。目的就是为了降低这三个模块的耦合度。

成员的管理：

• 交换机名称
• 队列名称
• binding_key
• durable

绑定持久化：数据恢复

绑定的持久化是利用sqlite3的指令，将数据写入到本地的磁盘中。主要是提供迅速的查询，能快速获取到全部的exchange---queue绑定。将获取的结果放到map<exchange,queue_map>中，形成交换机 ：队列 = 1 ：n的映射关系。大体上，绑定的持久化设计和交换机队列的基本一致。

对外提供的接口：表的创建，表的删除。提供数据库新增数据，移除数据。移除指定的交换机（必然要删除所有的指定交换机对应的queue）。移除指定的队列（移除某一个交换机对应的指定的队列）。数据恢复。

成员管理

• 数据库句柄

成员接口

• 表的创建/删除
• insert(exchange,queue) , remove(exchange,queue)
• 移除指定交换机 removeExchange(exchange)
• 移除指定队列 removeQueue(queue)

介绍一下Recover

Recover是通过数据库句柄的select获取到所有的交换机数据。在回调函数中，构建Binding的智能指针对象，通过map<exchange,queue_map>构建出引用的queue_map对象，然后将智能指针对象插入

（这样构建的好处是：如果exchange_map中对应的队列map不存在就会创建，如果是<exchange,queue>存在exchange的话 就会覆盖queue）

---

 绑定的管理：实际对外的接口

绑定的管理主要有：
绑定，在内存中新增一个交换机和队列的绑定关系（如果是持久化，在数据库中也要新增），对exchange_map的插入，在queue_map中插入。

解绑，解除指定交换机和队列的绑定（如果持久化，数据库中也要解除。）解除exchange_map中的指定queue；

移除交换机：对应的exchange_map中，删除指定exchange，对应的queue_map会自动删除。

移除指定队列，必须遍历所有的交换机，如果这个交换机有该队列，那么也要取消这条绑定。

获取某条关联。找到指定交换机，对应的exchange_map中的指定队列。

获取指定交换机的所有队列，以map<queuename,queue>的形式返回。

在重启的时候，能够进行数据恢复，就是在构建对象的时候调用Mapper的Recovery()

这些都涉及到多线程的访问，必须加锁！

成员的管理

• mutex
• 磁盘的数据管理句柄
• map<exchange,map<qname,queues>>

关于map<exchange,map<qname,queues>>的详细描述图

对外的接口：

• 绑定/解绑
• 移除交换机
• 移除队列
• 获取交换机的所有队列
• 获取某个绑定
• SIze()：绑定的总数
• Exists：绑定是否存在

在互斥锁下，实现的接口！

绑定和解绑

在互斥锁下，构建绑定的智能指针对象，通过引用exchange_map[exchange]，如果不存在对应的map<queue_name,queue_map>就会自动构建。接下来只要对引用对象insert(bind_ptr),如果设置了持久化，再往数据库插入。 

解绑，如果满足不存在交换机和队列的任意条件就直接退出。

获取交换机和绑定

因为是map<exchange,map<queue_name,queue>>对象，所以通过交换机和队列查找的时候，会涉及到许多的指针对象，要小心指针的越界访问。

---

消息的管理

消息来源于客户端的生产者，将消息推送到服务器上，再由服务器转发消息到客户端的消费者上。

消息是涉及网络传输的，必须考虑序列化和反序列化问题。我的设计是通过proto生成消息的属性，(消息的id,是否持久化，routingkey匹配规则)。

消息的管理是基于队列的，因为消费者会订阅某个队列，我们要将消息发给消费者，必然要获取队列。所以先有队列，才有消息。

消息是会先被转发到服务器上，在由服务器发送给订阅某个队列的消费者，其中消息会被短暂的保存在服务器上。考虑消息的存储一般是比较大的，所以不使用sqlite3存储，转而使用普通文件来存储消息。

另外如果在文件中存储的消息被ack后，就要将消息删除，删除的方式，并不是删掉指定的消息，然后把后续的消息往前挪。而是采用伪删除，即将当前的消息设置为无效位，等到文件中的无效消息达到一定条件的时候，就进行垃圾回收；

垃圾回收：会读取某个队列管理下的消息文件，如果读取到的是无效位，那么就不能加载该条消息到新文件，就应该把这条消息丢弃掉。等到新文件加载的全部有效消息之后，对文件重命名，并且删除旧文件。

考虑消息的存储格式

消息的存储就如果tcp传输层一样，我们要考虑粘包问题。

为了解决粘包问题：我的做法先计算消息body的长度，在文件中的存储保存它的lenth长度 接着存储body数据。在协议格式中存储它的总长度和偏移量，通过解析指定的消息就能拿到它的lenth和offset，读取它在文件中的位置。

 *  将消息在文件中管理,知道偏移量就能插入，考虑粘包问题(消息的长度)

 *  消息管理是基于对各自的队列管理，指定队列的名称,一个队列。如果持久化就创建一个文件管理,多个文件统一放在目录下(basedir)

消息的描述

• 消息的载荷：基本属性+body+有效位
• 偏移量
• 长度

消息在文件中的管理

 以队列是基础，创建文件保存消息的数据，需要队列名称。消息存储在哪一个文件中

MsgMapper的成员

• 队列名称  _qname
• 存储的文件_datafile

MsgMapper的接口

• 创建文件
• 向文件中插入
• 加载有效数据
• 垃圾回收

总的设计思路

1.创建目录，创建指定文件。

 2.往指定的文件里，插入一条消息,插入先插入4字节消息的长度，再插入消息body(序列化后),最后更新

3.消息的删除,只需要将覆盖写，将消息上的有效位转为无效

4.垃圾回收机制:总消息超过2000，有一半消息无效

         1)获取所有的有效消息 2)往临时文件里写数据 3)更新消息的偏移量 4)删除源文件，对临时文件重命名

由于这个模块是比较复杂的，这里就仔细介绍一下

插入数据

向文件中插入数据，先获取序列化消息的body长度，获取文件的偏移量，先将8字节的文件长度写入 文件中，在将序列化的消息数据写入文件。最后通过智能指针更新消息的偏移量，offset（位8字节lenth长度的起始位置），lenth为文件的body长度。

加载文件中的数据 

 从文件头一直读取到文件尾部，读取的时候，先读取8字节（size_t)的长度，接着读取序列化的文件，将偏移量移到下一个8字节的起始位置。构建消息，将读取到的body反序列化。最后将消息放到链表中。继续读取下一轮的消息。

为什么考虑链表，主要是链表相对于队列适合访问中间位置的消息。

 

垃圾回收 

垃圾回收就是有效位为"0"的消息删除，在文件中删除是涉及到挪动数据，很麻烦。我们的做法是创建临时文件，往临时文件里插入有效消息。最后将临时文件重命名为数据文件，并且删除旧文件。由于涉及到文件的IO，效率是很低的，所以垃圾回收前要先进行一次check,只有当消息超过2000条并且有效消息低于50%，才进行垃圾回收，这个检查是由上层来做的。

步骤

• 1.加载所有有效消息
• 2.调用Insert（）
• 3..rename临时文件
• 4.重命名

 以队列管理消息

以队列管理消息，需要队列名称作为标志。另外消息是涉及是否持久化的，就需要在文件中的管理句柄。为了能够记录持久化的消息有哪些，通过durable_map快速定位到智能指针。

当消息推送给消费者之后，消息就会进入待ack状态，等待客户端的ack，来决定是否将数据消息删除，如果一直没有收到客户端的ack，就会再次发送。所以我的设计中，所有推送的数据会被放到wait_ackMap中。

对map的访问是涉及多线程的，存在线程安全。必然要添加互斥锁保护。

队列的成员：

• mutex
• 文件管理句柄mapper
• 队列名称qname
• 持久化map
• 待确认Map

以队列管理消息，提供的接口应该有：

• 数据恢复：服务器重新加载的时候，保存在文件中的消息应该被重新加载到内存中。
• 垃圾回收，清理文件中的无效消息。
• 新增消息，在队列中增加一条消息。
• 推送消息：即获取队列头的数据，并且添加进等待ack队列。
• 删除消息：删除一条消息，如果是持久化，就要连同文件中的一起删除。

数据恢复

这里巧妙的设计，如果我想进行数据恢复，那么我会调用一次mapper的垃圾回收，因为mapper的垃圾回收返回值就是一个带智能指针的链表，获取到所有的智能指针后再插入到内存中。

管理消息队列 ：实际对外接口

消息的管理是以队列为基础，在内存中必然会存在很多消息队列，必须将这些队列管理起来。管理的方式依旧采用map。以qname为first,shared_ptr<qname>为second，对外提供初始化消息队列，即如果有队列创建，那么必须整合消息，创建出消息的文件，消息的句柄等等。

删除队列。如果一个队列被删除，连带的消息队列也需要被删除。

获取某个队列上的消息。

往某个队列上新增消息

手动ack：主动移除队列上的某条消息

MessageManager的成员

• basedir:消息文件的目录
• map<qname,queuePtr>
• mutex：访问map涉及到多线程

MessageManager的对外接口

• InitQueue()
• Insert()
• DeleteQueue()
• GetFront()
• Ack()

基于队列的管理是比较简单的，属于常规的操作，调用map和queuePtr的函数接口等。

基本的步骤就是添加互斥锁，map中查找queuePtr，调用queuePtr的函数接口。

例如：新增消息

 

---

虚拟机——模块整合

虚拟机是对交换机、队列、绑定关系、消息等模块的统一管理。上层只需要创建虚拟机就能调用交换机、队列、绑定、推送/消费消息。实际上就是对之前的模块进行封装。虚拟机的模块是比较简单的，实现起来就是一些接口的附用。

再来回忆一下交换机、队列、绑定、以及消息之间的关系

以虚拟机为主体的服务端，主要负责路由和推送功能。推送的方式是多样的，可能有主题、直接、广播推送。就决定着生产方将数据逐个推送到每一条队列是一个费时、费力的过程。因此消息是先交给交换机，由交换机根据推送到哪一条队列进行推送。

所以建立绑定关系，将队列和交换机先建立连接。

消息的存储是以队列为单元的(存/取方便)。

虚拟机的描述

设计的思路，要能声明/删除交换机，声明/删除队列，建立交换机和队列的绑定关系/解除绑定关系，消息的推送和消费 ，消息的删除

成员：

• 交换机管理句柄
• 队列管理句柄
• 绑定管道句柄

对外接口

• 声明/删除交换机
• 声明/删除队列
• 绑定/解除绑定
• 推送和消费
• ACK

构造：

在创建交换机的时候，会进行交换机、队列、绑定的数据恢复，但是消息必须先初始化队列数据，因此要先对队列的获取。

需要注意的是队列的创建后，还需要对消息队列初始化

值得注意删除不仅是单个模块的删除，还需要解除绑定关系，如果删除队列，就要清空队列的消息。

生产，直接附用消息管理句柄的插入，将数据放到队列中。

消息，获取头数据，并且放进确认队列

---

路由模块----功能类

之前谈到如果生产出的数据全部由自己发送到队列，会造成效率低下，具体体现在。如果有多个 队列同时订阅了这份数据，那么就需要发送n次。因此生产出来的数据是先交给交换机，根据路由匹配规则（直接发送，主题发送，全广播）将数据发送到指定的队列上。

消息持有routingkey,routingkey是由“.” - "a~Z" "_" "0~9"组成的字符串。形如news.music.pop。

队列持有bindingkey，由"#" " * " "_" "." "a~Z" "0~9" 组成的字符串。形如new,music.#.pop

匹配规则：#匹配一个或者多个字符 ，*匹配一个字符  .作为字符的分隔符，不参与匹配

---

路由模块是一个功能模块主要功能是判断routingkey的合法性，判断bindingkey的合法性

路由函数：根据交换机的类型，以及匹配规则，判断是否要发送。

因为存在#这个特殊的字符，它能够匹配一个或者多个字符串，如果我们简单的比较，就会是n^2的复杂度，常见的优化就是利用动态规划用一个dp[m][n]的空间换取O(N)的时间。

下面详细介绍一下这个动态规划

比较字符串是否相等如果当前字符串相同，则判断对角是否为true，不为true则比较失败。

如果遇到#，#号可以匹配任意个字符，也就是说dp当前位置可以取决于左边和对角

注意初始化问题，为了防止越界，dp的存储m+1 ,n+1

dp[0][0]位置是true 是为了比较routingkey[0]  和 bindkey[0]的元素

如果bindkey[0]如果是# ,意味着可以匹配routingkey的任意数量字符，就要将dp[0][i]全部置为true

画图举例：比较aaa.bbb.ccc.ddd    aaa.#.ddd

            std::vector<std::string> rkeys, bkeys;
            StrHelper::splite(routing_key, ".", rkeys);
            StrHelper::splite(binding_key, ".", bkeys);
            int m = rkeys.size(), n = bkeys.size();
            std::vector<std::vector<bool>> dp(m+1, std::vector<bool>(n+1, false));
            // 初始化
            dp[0][0] = true;
            if (bkeys[0] == "#")
            {
                for (int i = 1; i <=n; i++)
                {
                    dp[0][i] = true;
                }
            }

            for (int i = 1; i <= m; i++)
            {
                for (int j = 1; j <=n; j++)
                {
                    if (rkeys[i - 1] == bkeys[j - 1] || bkeys[j - 1] == "*")
                    {
                        dp[i][j] = dp[i - 1][j - 1];
                    }

                    else if (bkeys[j - 1] == "#")
                    {
                        dp[i][j] = dp[i - 1][j] | dp[i][j - 1] | dp[i - 1][j - 1];
                    }
                }
            }
            return dp[m][n];

---

对外提供的接口：

• routingkey的合法
• bindingkey的合法
• route

路由模块的涉及是比较简单的，这里就不详细介绍了，结合动态规划的代码是很轻松就能实现功能的。

消费者

上面中，消息的管理是基于队列的管理，而在队列中有数据之后，就要将消息推送给消费者。

消费者的管理问题

消费者是众多的，需要进行管理。考虑到连接问题，在我的设计中，最小连接单元是一个信道，是将connection再做细分的单元。而一个消费者必然对应一个消费者，考虑将消费者用信道管理，这样的好处是。如果信道关闭了，方便清楚对应的消费者。但是在推送中的问题，如果需要从队列中将数据发送出去，就需要先建立信道，是比较麻烦的。所以将以信道为管理，转变为以队列进行管理。

消费者的描述

消费者属于的队列名称qname，消费者自身的标记tag，是否自动删除(将消息推送给消费者后，是否立即将消息删除)，回调函数（用于消费者处理任务）。

• qname;
• tag;
• callback;
• auto_delete

消费者在队列中的管理

设计思路

我们希望以队列管理消费者，所以队列必须能够新增一个消费者，删除一个消费者。

RR轮转负载均衡选择

选择一个消费者，我们的设计是如果队列上有消息，我们会通知一个消费者来接收消息，并不是全部推送，所以就需要负载均衡的选择，利用自增+取模，得到一个轮转的数，这就是RR轮转的思想。然后根据这个数字去获取一个消费者。

队列管理消费者的成员

• _mutex：因为涉及到多线程，多线程会对某一个队列同时操作
• _qname:指定队列
• _rr_sql:轮转数字
• vector<消费者>：根据下标能获取消费者

对上层提供的操作

• Create()
• remove()
• choose()

消费者模块也是比较简单的，这里就不过多的介绍

---

消费者队列的管理

在项目中，我的思想是会存在很多个不同队列，每一个队列都存在消费者，所以就会要对这些队列管理。管理是通过map<qname,map<tag,消费者>>的管理，思路和消息的管理有点相似。

另外对消费者队列的操作是涉及多线程的，需要加锁保护，我们基于队列的设计来考虑消费者队列管理的对外接口。这些接口应该有：初始化一个队列，如果有一个新的队列到来，我们就将他加到map中。创建/删除消费者，往map中的second添加/删除消费者。智能选择。负载均衡的选择某一个队列的一个消费者。

管理的成员：

• _mutex
• map<qname,map<tag消费者>>

对外提供的接口

• 初始化队列：不存在就新增
• 新增消费者：根据队列名称获取到句柄，往队列句柄中添加数据
• choose:根据队列的句柄，选择数据

同样，由于这个管理是比较简单的，就不进行过多的介绍。通过图理解这一个逻辑

用队列管理消费者：
本质也是一个map，以消费者标志的为key，消费者的智能指针为值的存储。

在map中管理消费者队列

---

信道：最小的连接单元

如果将一个connection作为一个客户端和服务器的连接，那么消耗就比较大。因此，将connection再次细分为多个信道，由这些信道整合构成connection。

信道的设计是比较麻烦的模块，首先信道可以看成是连同一个生产者和消费者的（但是实际上信道的中间必然会经过服务器的业务处理）。另外信道涉及网络传输，我们会通过自定义协议的方式解决消息传输过程中序列化反序列化和粘包的问题。

站在生产者的视角上，生产者想要往队列上放数据，会先通过probuf进行序列化和反序列化，解决粘包问题是通过muduo库的配合完成。消息到达服务器会先交给交换机，交换机根据路由规则，派发到相应的队列上。然后消息组织起来，传入回调方法。因为网络传输是极大消耗时间的IO型，所以我们采用线程池，将数据丢到线程池子中。至此生产者的任务就结束了。

再来根据消费者的视角，线程池里面有数据，就会立马取出信道对应的队列的消息和消费者（这里展示了将消费者和消息以队列管理的好处）。回调消息处理的方法，将消息发送到对应信道的客户端上，客户端就能读取消息了。如果消费者是自动删除的，那么就不需要等到ACK就将消息删除掉。

这里就衍生出来俩种消费消息和订阅消息，也是本文的难点，在下文中会进行详细的介绍。

信道的描述

信道的唯一标识，这个信道对应的消费者，管理消费者的句柄（方便在断开信道时删除消费者）。

connection和codec分派器。线程池负责分派任务，虚拟机（队列、交换机、消息等的整合）

信道的成员:

• cid:信道标记
• _consumer:消费者
• _map<consumer_name,consumer>信道管理句柄
• _conn(Tcp连接器)
• _codec(分派器)
• virtual(虚拟器)
• threadpool（线程池）

信道的管理

信道提供的接口：

• 声明/删除交换机
• 声明/删除队列
• 绑定/解绑
• 推送
• 订阅队列
• 消费消息

声明删除绑定都是调用虚拟机的接口，再往服务器发送基本请求。是一个简单的过程。

消息的推送

推送消息的比较复杂的，下面详细介绍一下，推送是由生产者的请求通过交换机路由，在推送到交换机绑定的队列上。必须先获取指定交换机，接着根据路由匹配规则，将消息投递到指定的队列当中。

consum消息的消费

consum是线程池的任务，当有数据被送到队列中，就会自动执行

基本思路就是通过队列名称往虚拟机里取出一个任务，在取出一个消费者，执行消费者的回调函数，同时如果消费者设置了自动删除，就无需等到客户端的回应，就立即将消息删除。

订阅队列 

订阅队列就是创建消费者，再设计一个消费者的接口（这里我们统一将消费者接口设计为简单转发功能，再将消息发送到对应信道的客户端上），再调用基本的响应。

 

执行任务的回调函数，会回调到消费者绑定的callback将消息发送到对应信道的客户端，最后是否自动删除。 否则，我们会手动进行删除。

信道涉及到网络的通信，是不好测试的，等到后面会进行服务器的功能联调。

值得注意的是

• 在声明队列的时候，必须进行消费者初始化，即对消费者进行添加队列。
• 在删除队列的时候，对应的消费者必须清空。

---

Connection连接模块

最小的连接单元是信道，连接是对客户端的响应，再调用信道的管理句柄，创建信道该有的服务。如：打开信道、关闭信道、获取一个信道。

对连接的管理，实际上就是对信道的增删查。创建信道余需要有Tcp连接，codec协议处理，异步线程池，消费者管理句柄。这些是外对connection提供的参数。

成员：

• tcpconnection
• codec协议处理器
• 消费者管理句柄
• 虚拟机管理句柄
• 异步线程池
• 信道管理句柄

对外接口：

• OpenChannel
• CloseChannel
• GetChanel

这些设计的主要思想就是调用channel管理句柄的对应方法，然后向客户端发起基本请求。

属于是简单的操作，由于篇幅有限就不进行过多的介绍。

ConnectionManager---实际对外接口

连接的管理，实际上就是将tcpconn和Connection类用Map管理起来，然后对外部提供conn的增加删除获取，但是这里对map的访问是涉及多线程的，需要加锁的保护。

---

Broker整合

Broker是一个epoll模型

broker是一个服务器任务注册和派发模块。

基本流程是broker接收到客户端的请求，根据codec协议处理器，进行解包和反序列化，获取对应的请求。dispatcher事件派发器拿到指定的请求进行分发。根据提前为请求注册的事件执行任务。

所以我们的设计很简单：
首先需要muduo库的tcpconn和evenloop，协议处理器codec，请求派发器dispatcher，总体将框架搭建起来。在往上层关心用户需要哪些请求

总体的请求应该有：

• 创建/删除连接
• 打开/关闭信道
• 声明/删除交换机
• 声明删除队列
• 绑定/解除绑定
• 基本推送/消费
• 订阅/取消订阅队列
• Ack

由proto为我们生成请求的类型，简单说明一下：muduo协议存储采取的是L-V模式，即总体L表示len长度，对应的每一个请求都是len+val的模式。解决了数据粘包的问题。

因此为了满足上层业务的处理，需要提供的成员应该有:连接管理句柄提供打开信道和关闭信道，虚拟机管理句柄设计交换机、队列和绑定、消息的关闭，消费者的管理句柄。

broker对应的成员：

• tcp连接
• evenloop主事件循环：不断进行IO、派发事件
• codec:协议处理__muduo库是采用l-v处理协议的
• dispatcher事件派发：根据请求派发到注册的方法
• virtual_host：虚拟机的管理
• connection连接的管理
• consumer消费者的管理句柄

关于broker的注意点:

在构造的时候，虚拟机会进行数据恢复，涉及交换机队列等等。而消费者必须也进行初始化，它的初始化是以队列为核心的，所以必须先获取存储的所有队列的句柄然后提供给消费者进行Init。

在为请求提供对应的方法的基本格式应该是：1.判断连接是否存在，不存在就中断发送错误请求。 2.判断信道是否开启，不开启就关闭，发送错误请求。 3.调用channel的操作后，发送基本请求。

以队列为例，展示基本格式：

下面以一张图快速了解broker的处理过程

---

 至此服务器的搭建就完成了，涉及到的模块整合还是挺多的，它们中最大的亮点就是强解耦和模块的完整性，明显的先描述在组织，一层一层往上搭建的模块。

下面来介绍一下客户端。

客户端主要涉及消费者的描述，有消费者自己提供处理消息的方法。

信道的管理，服务器中，信道是最小的连接单元，服务器需要对信道发送请求，交到服务器中。

连接管理，实际上对应服务器的协议处理和事件派发。

线程池模块，主要分为连接线程池和异步线程池，连接线程池主要处理网络IO，异步线程池处理事件的回调，任务的处理等等。

这四个模块就组成客户端，客户端的编写还是比较容易的。

---

消费者的描述

对于一个消费者，应该具有的是消费者标签，标志是哪个消费者。

队列名称，标记消费者订阅了哪一个队列。

是否自动删除，客户端推送出消息后，是否自动将消息删除，而不是等到客户端的ack。

业务处理函数，为消费者绑定处理的方法，当消费者接收服务器推送而来的消息时候，如何对消息处理。

消费者的成员

• tag：消费者标签
• queue_name
• auto_delete
• callback：业务处理函数

---

信道模块

客户端的信道模块和服务器的信道模块略有区别。信道是连接的最小单元，对连接进行细分。原则上服务器对于信道设置什么请求，客户端的信道就应该有什么请求。例如：

• 声明/删除交换机
• 声明/删除队列
• 绑定/解绑队列
• 生产/消费
• 订阅/取消订阅
• ack消息

如果一个客户端需要向服务器发送请求，例如客户端请求声明交换机，那么在服务器上的请求必须是被执行完成而且是成功的，客户端才能进行下一步。而客户端对服务器的这一步确认是必要，所以我们的设计是每一个服务器对请求完成后，都会发送基本响应（包含一个ok标签，cid信道id,rid请求id）。客户端接收到请求后，会以rid为键值，将数据保存起来。

必须等待确认才能进行下一步操作，这是同步机制。关于同步，声明时候唤醒线程执行？

当客户端向服务器发送请求，请求得到服务器的ack后，将ack放到客户端的响应map后，就可以唤醒线程。

这里又涉及到关联，需要收到服务器发来的ack，并且将ack保存起来，这一个存放响应的操作。

我们想这样涉及：当客户端收到协议报文时候，会进行解包，如果是ack回应，就会调用信道的putAck方法，将请求放到map中，同时阻塞住的线程要检测对应的ack是否存在于map中，如果存在就唤醒线程。

对于请求map的操作是涉及多线程的，需要互斥锁的保护。

所以我们这里在设计俩个接口

1. putAck:将ack放到map中，并唤醒所有的线程。
2. waitAck:阻塞等待，直到有对应rid的ack发送过来。

信道的成员：

• 信道id
• 协议处理器codec
• tcp连接
• 消费者：这里设计的是，一个消费者对应一个信道
• mutex:互斥锁
• condiction_variable
• map<rid,响应>

对外提供的接口：

• 声明/删除交换机
• 声明/删除队列
• 绑定/解绑队列
• 生产/消费
• 订阅/取消订阅
• ack消息
• PutAck
• WaitAck

对外接口的基本思想就是创建一个唯一的请求rid,构建请求，发送请求，阻塞等待服务器的ack

总体来说设计是比较简单的，就拿发布消息举例

1. 请求rid
2. 构建请求
3. 发送请求
4. 阻塞等待ack

了解一下阻塞等待的写法

        BasicCommResponseptr WaitResponse(const std::string &rid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cv.wait(lock, [this, &rid]()
                     { return _response.find(rid) != _response.end(); });
            BasicCommResponseptr resp = _response[rid];
            _response.erase(rid);
            return resp;
        }

这是一种很常见的等待方法，利用lambda和条件变量的结合。将同步变量设置为while(1)的死循环，直到条件成立。

信道对外的实际接口：
简单来说，一个连接可能有很多的信道，那么就需要对这些信道进行管理。管理是利用map的方式，以信道id为键，信道的智能指针为value。对外提供简单的增加/删除/查找信道。

需要注意的是对map的操作是涉及到多线程访问，存在线程安全的，需要加锁保护。

---

连接：Client服务器

实际上的客户端服务器就是连接，整合信道的管理。进行连接的管理和事件的派发。和服务器的broker是差不多的。在连接中主要是为dispatcher注册方法，客户端会根据请求进行事件的派发。

如果是网络IO型的，会添加到muduo库的线程池中，会将任务放到异步线程池中，这样设计的好处是用户不需要自己选择哪个线程池，只管调用即可。

连接的成员：

• tcp连接
• client
•  muduo::CountDownLatch _latch:同步计数器
• 协议处理器codec
• 事件派发器dispatcher
• 工作线程池：网络IO线程池和异步事件线程池
• 信道的管理句柄

对外提供的方法

• 打开信道
• 关闭信道

关于注册的方法是值得注意的

消费任务

异步线程池里存放任务

调用信道的消费任务方法，信道会回调消费者的任务。

存放基本响应

当客户端接收到服务器的ack后，由派发器回调方法，就要立即执行信道的请求存在在map中，信道会通知/唤醒所有的线程，那么如果有线程在阻塞，就会根据条件判断是否解锁，执行下一步操作。所以与上文的同步关系结合，高效的体现。

总体上客户端服务器的搭建就结束了，总的来说就是封装了消费者、线程池、信道、连接、最后将这些利用muduo库的dispather事件派发器、codec协议处理器整合成为客户端。

梳理一下客户端的逻辑

站在生产者的角度上。生产者先于服务器建立连接，期间会阻塞等待，直到连接成功。

建立连接后会声明交换机，队列，绑定关系。这些都是客户端向服务器发送请求在服务器上完成的，完成之后服务器会向对应的信道发送rid的基本报文，客户端服务器接收到响应后codec协议解析，获取响应的类型，dispather派发器会根据响应的类型，调用响应的方法，因为是基本ack响应，会将响应放到响应map中，并且唤醒所以的阻塞去比较rid是否一致。如果是相同的，就会进行下一步的操作。

直到建立完成绑定关闭之后，生产者客户端回向服务器发送消息，服务器会进行转发到交换机，交换机有队列后，会根据bandingkey和routingkey的匹配规则，将消息推送到指定的队列上。如果存在消费者，就会立即将消息发送到消费者客户端上。

消费者客户可以主动订阅某一个队列，它的操作和生产者差不多，先与服务器建立连接，声明交换机，队列，绑定关系，向服务器发送订阅队列的请求，等到服务器的ack。之后就是如果服务器发送消息到消费者处，dispatcher会调用消费者的业务回调函数，处理业务。

---

功能联调

梳理一些功能联调时候遇到的问题

错误主要分俩类：逻辑错误和程序异常终止

逻辑错误主要是达不到预期的结果，比如客户端一直连接不上服务器，消费者添加失败，服务器中不到消费者的管理句柄，服务器接收不能主题发送。

程序异常终止主要发生了段错误，和aborted，都是进程被信号杀掉。

出现段错误等错误是比较好调试的，很快就能找到问题存在，但是逻辑问题是比较麻烦的，有时候就会找了很久。

下面举几个经典的例子说明一下，我遇到的问题。

段错误

消费者不存在导致的段错误。我通过gdb调试，在查看调用堆栈，找到在程序的某一行出错后，连续跳转几层，发现经常是对空指针的解引用。在未找到消费者时候，没有退出。

因为在我的项目中，用了大量的智能指针，并且对指针智能重名， 很多时候会把make_shared<>写成shared_ptr导致会对指针解引用

但是这类问题熟悉一些gdb是很好找到的。

逻辑错误

连接有问题，

在dispatch中发现问题是在服务器连接的时候，没有调用自己创建的connectionOpen

---

消费者订阅队列后

生产者客户端发送数据，服务器中不到 消费者

排查：

• 首先猜测生产者是否将数据发出，然后将数据打印处理
• 检查服务器是否接收到数据
• 检查消费者订阅是否存在问题
• 发现消费者是存在的
• 最后找到问题是，在channel中订阅了消费者，并没有将channel中对应的消费者更新，导致一直找不到消费者

---

找不到消费者的管理句柄

 先检查了消费者的创建是否出现问题，再去从客户端的请求出来，跳到队列的声明，然后去消费者订阅模块检查，发现问题是在订阅请求的时候，在队列初始化的时候，没有将消费者初始化，并且添加到channel的consumer当中。解决的办法主要是打印。

---

routingkey发送一直错误

我发现routingkey一直有问题，但是像消息的rid 和 cid 以及body等都是没问题的。所以我就想服务器在broker接收到请求的时候，是否有问题。先把报文打印出来看看，然后是没问题的。我又去看了客户端发送是否存在问题，我在send之前打印出消息，发现消息的属性并没有被真正设置进请求，因为它是嵌套的proto定义，正确设置了之后，就解决了这个routingkey找不到的问题。