前面的五篇文章我们讲解项目的环境搭建以及项目的技术前置接下来开始我们真正的项目概述环节一.需求分析1.核心概念生产者 (Producer)消费者 (Consumer)中间人 (Broker)发布 (Publish)订阅 (Subscribe)一个生产者 一个消费者N 个生产者 N 个消费者其中Broker Server 是最核心的部分, 负责消息的存储和转发。而在 AMQP(Advanced Message Queuing Protocol-高级消息队列协议一个提供统一消息服务的应用层标准高级消息队列协议为面向消息的中间件设计使得遵从该规范的客户端应用和消息中间件服务器的全功能互操作成为可能)模型中也就是消息中间件服务器 Broker 中又存在以下概念虚拟机 (VirtualHost): 类似于 MySQL 的 database, 是一个逻辑上的集合。一个BrokerServer 上可以存在多个 VirtualHost交换机 (Exchange): 生产者把消息先发送到 Broker 的 Exchange 上再根据不同的规则, 把消息转发给不同的 Queue队列 (Queue): 真正用来存储消息的部分 每个消费者决定自己从哪个 Queue 上读取消息绑定 (Binding): Exchange 和 Queue 之间的关联关系Exchange 和 Queue 可以理解成 多对多 关系使用一个关联表就可以把这两个概念联系起来消息 (Message): 传递的内容所谓的 Exchange 和 Queue 可以理解成 多对多 关系, 和数据库中的 多对多 一样. 意思是:一个 Exchange 可以绑定多个 Queue (可以向多个 Queue 中转发消息)一个 Queue 也可以被多个 Exchange 绑定 (一个 Queue 中的消息可以来自于多个Exchange)对应到业务图理解起来就是要实现的内容broker 服务器消息队列服务器消息发布客户端向服务器发布消息消息订阅客户端从服务器订阅消息AMQP 协议中细化了实现规则交换机队列绑定虚拟机VirtualHost虚拟主机上述数据结构 既需要在内存中存储 也需要在硬盘中存储内存存储 方便使用硬盘存储 重启数据不丢失2.核心 APIBroker 服务端需要提供一组核心 API依靠这些接口完成消息队列的基础能力创建交换机exchangeDeclare删除交换机exchangeDelete创建队列queueDeclare删除队列queueDelete建立交换机与队列的绑定queueBind解除交换机与队列的绑定queueUnbind发布消息basicPublish订阅消息basicConsume消息消费确认basicAck取消消息订阅basicCancel生产者Producer与消费者Consumer通过网络远程调用 Broker 的上述 API以此实现生产者 - 消费者模型。关于 VirtualHost对于 RabbitMQ 来说, VirtualHost 也是可以随意创建删除的此处咱们暂时不做这部分功能(实现起来也比较简单, 咱们的代码中会完成部分和虚拟主机相关的结构设计大家可以自行完成管理逻辑)3.交换机类型RabbitMQ 一共提供四种交换机类型Direct、Fanout、Topic、Header。 Header 交换机使用复杂日常开发很少用到本项目只实现前三种常用交换机。Direct 直连交换机生产者发送消息时直接指定交换机所绑定队列的名称。Fanout 扇出交换机消息会被复制分发到该交换机绑定的全部队列。Topic 主题交换机队列绑定交换机时设置bindingKey发消息时携带routingKey。当两个键满足通配匹配规则消息就投递到对应队列。用 QQ 群红包来类比理解Direct专属红包只有指定的人可以领取。Fanout全员红包发一次红包群内所有人都能收到这份红包。Topic答题 / 画图红包红包附带条件满足条件的成员才能领取红包。4.持久化交换机、队列、绑定关系、消息都需要持久化存储。 实现持久化后当服务进程或者服务器主机重启这些数据不会丢失。5. 网络通信生产者、消费者属于客户端程序Broker 作为服务端双方通过网络交互。 客户端需要提供一系列 API远程调用服务端能力创建 Connection关闭 Connection创建 Channel关闭 Channel创建队列queueDeclare删除队列queueDelete创建交换机exchangeDeclare删除交换机exchangeDelete创建绑定queueBind解除绑定queueUnbind发布消息basicPublish订阅消息basicConsume消息确认basicAck取消订阅basicCancel对比服务端 Broker 接口可以发现客户端额外增加了 Connection 与 Channel 相关操作Connection底层对应一条 TCP 连接ChannelTCP 连接内部的逻辑通道一条 Connection 可以持有多个 Channel各个 Channel 之间数据相互隔离互不影响。该设计的核心目的是复用 TCP 长连接减少频繁创建、销毁 TCP 连接带来的开销。通俗理解Connection 相当于一根网线Channel 就是网线内部独立的多条信号线。6.消息应答消息被消费后需要执行应答分为两种应答模式自动应答消费者拿到消息即视为消费完成Broker 直接删除这条消息。手动应答消费者业务处理完成后主动调用应答 APIBroker 收到应答请求才会删除消息。手动应答用来保障消息被消费者真正处理完成在高可靠业务场景中广泛使用。二.模块划分消息队列系统模块划分整套消息队列系统分为三大组成部分服务端发布客户端订阅客户端服务端模块1.交换机数据管理a.交换机需要维护的数据字段交换机实体保存用于描述自身属性的相关数据各字段作用如下1.交换机名称交换机的唯一标识用来区分不同交换机。2.交换机类型决定消息路由转发逻辑。队列绑定交换机时会设置binding_key生产者发布消息时携带routing_key依靠这两个键完成路由判断Direct直接交换消息的routing_key和队列绑定的binding_key完全相等消息投递至该队列。Fanout广播交换无视routing_key消息会投递到当前交换机绑定的全部队列。Topic主题交换按照通配符规则匹配routing_key和队列的binding_key匹配成功就投递消息。3.持久化标志布尔标记用来标识交换机元数据是否写入磁盘持久化。开启持久化后 Broker 重启交换机信息不会丢失。4.自动删除标志布尔标记。当所有与该交换机建立关联的客户端全部断开连接后是否自动删除这个交换机。5.交换机其他参数预留扩展字段本项目暂不使用。b.交换机对应的管理操作接口创建交换机exchangeDeclare采用 AMQP 声明式强断言思想如果同名交换机已经存在则直接复用不报错不存在则新建交换机。删除交换机exchangeDelete一个交换机可以绑定一个或者多个队列对应多条绑定记录。删除交换机时需要级联删除该交换机下所有的绑定关系。获取指定名称交换机根据交换机名称查询并返回对应的交换机实例信息。获取当前交换机数量统计服务端内一共存在多少个交换机2.队列数据管理a.需要管理的数据字段队列名称队列的唯一标识用于区分不同队列。持久化存储标志布尔标记控制队列元数据是否持久化保存。开启持久化后Broker 重启队列仍然存在。是否独占标志独占模式开启之后仅允许当前这一个客户端订阅该队列消息其他客户端无法订阅。自动删除标志当所有订阅该队列的客户端全部断开连接后是否自动删除队列本项目暂未实现。其他参数预留扩展字段暂未使用。b.提供的管理操作核心就是对队列数据做增、删、查操作创建队列queueDeclare删除队列queueDelete获取指定队列信息根据队列名称查询队列属性。获取队列数量统计当前服务端一共有多少队列。获取所有队列名称服务重启时需要重新加载历史数据与消息消息按队列维度存储存储文件以队列名称命名。加载消息前必须先拿到全部队列名称。c.持久化关联规则如果队列的持久化标志为 falseBroker 重启之后队列本身就会消失没有客户端可以订阅该队列。 这种情况下就算把队列内消息持久化保存下来也没有意义。因此约定队列非持久化时该队列的消息也不需要持久化。3.绑定数据管理模块绑定模块用来记录队列与交换机之间的关联关系。a.管理的数据字段交换机名称绑定关系所属交换机的标识队列名称绑定关系所属队列的标识binding_key绑定密钥用于 Direct 直接交换机、Topic 主题交换机的消息路由匹配规则。 支持由数字、字母、下划线_、#、.、*组成。示例 binding_keynews.music.#消息携带的 routing_keynews.sport.footballb.管理的操作添加绑定建立交换机和队列之间的绑定关系。解除绑定删除交换机与队列之间的绑定关系。获取交换机相关的所有绑定信息删除交换机时需要级联删除该交换机下全部绑定记录消息投递至交换机时依靠绑定信息判断消息要投递到哪些队列。获取队列相关的所有绑定信息删除队列时需要级联删除该队列对应的全部绑定记录。获取绑定信息数量统计当前所有绑定关系总数。4.消息数据管理a.消息信息消息属性:ID消息的唯一标识。持久化标志标记该消息是否需要持久化最终是否落地还依赖队列的持久化标志。routing_key消息路由键。消息发送到交换机后依靠该字段与绑定关系的binding_key匹配决定投递到哪些队列。消息主体消息的实际内容。服务端管理附加信息存储偏移量消息以队列为单位保存在文件中偏移量代表当前消息相对于文件起始位置的字节位置。消息长度从偏移位置读取指定长度的数据用来解决粘包问题。是否有效标志标记消息是否已被删除。删除消息不会直接前移拷贝后续数据仅修改该标志当文件内有效消息占比低于 50%且总消息数量超过 2000 时触发垃圾回收重新整理消息存储。系统重启时只需要重新加载标记为有效的消息等效于执行一次垃圾回收。b. 消息的管理管理方式以队列为单位进行消息管理所有消息相关操作都基于队列。管理数据:消息链表保存所有待推送的消息。待确认消息 hash消息推送给客户端后等待客户端返回消费确认收到确认后才真正删除消息。持久化消息 hash假设消息开启持久化存储垃圾回收会改变消息在文件中的存储位置。内存中记录消息实际存储位置垃圾回收完成后必须更新持久化消息对应的位置信息。垃圾回收逻辑读取全部有效消息重新截断文件把有效消息连续写入新文件。持久化的有效消息数量持久化的总的消息数量用于判断触发垃圾回收的时机。管理操作:向队列新增消息获取队首消息取出消息后将消息从待推送消息链表移除转入待确认消息集合。对消息进行确认从待确认消息集合移除消息并在持久化存储中删除该消息。恢复队列历史消息在队列对象构造函数中执行仅 Broker 重启时触发。垃圾回收由消息持久化子模块完成持久化文件中有效消息占比小于 50%并且总消息数量大于 2000执行垃圾回收。删除队列相关消息文件队列被删除后队列对应的消息文件不再保留。c.队列消息管理初始化队列消息结构移除队列消息结构队列创建、删除时调用。向队列新增消息对队列消息进行确认恢复队列历史消息5.虚拟机数据管理模块虚拟机数据管理模块是对交换机、队列、绑定、消息这四类数据管理能力的整合封装。a.要管理的数据交换机数据管理句柄队列数据管理句柄绑定信息数据管理句柄消息数据管理句柄b.要管理的操作声明 / 删除交换机注意删除交换机时需要级联删除该交换机对应的全部绑定信息。声明 / 删除队列注意删除队列时需要级联删除对应的绑定信息以及队列下所有消息数据。队列的绑定 / 解除绑定注意执行绑定操作前必须保证对应的交换机和队列都已经存在。获取指定队列的消息对指定队列的指定消息进行确认获取交换机相关的所有绑定信息消息投递至交换机时交换机会读取全部绑定信息根据路由规则判断消息应该投递到哪些队列。6.路由匹配模块路由匹配模块的作用判断一条消息能否投递到对应的队列。在队列与交换机的绑定记录中存在binding_key代表队列侧设置的匹配规则。 待发布的消息自身携带routing_key代表消息的发布标识。交换机分为三种类型广播、直接、主题。广播类型将消息直接推送给绑定在该交换机上的全部队列。直接类型routing_key和binding_key完全相同才算匹配成功。主题类型binding_key作为匹配规则例如news.music.#消息的routing_key例如news.music.pop满足规则则匹配成功消息才能投递。路由匹配模块不需要持久化管理数据对外只提供路由匹配相关操作提供接口用于判断routing_key与binding_key是否匹配。校验routing_key合法性仅允许数字、字母、下划线_、点.。校验binding_key合法性仅允许数字、字母、下划线_、点.、#、*。7.Broker 服务器模块Broker 属于功能整合模块该模块本身不对外提供具体业务操作核心作用是整合各类资源作为资源统一载体。单个服务器内置一个工作线程池所有信道相关操作复用这同一个线程池。单个服务器内置一个虚拟机交换机、队列、绑定、消息的全部操作都基于该虚拟机完成。单个服务器内置一套消费者管理模块。通信层的连接管理、协议处理模块句柄整个服务器共用一套。客户端模块1.消费者管理模块客户端分为两类消息发布端、消息订阅端。只有订阅了指定队列消息的客户端才被称为消费者。a.消费者数据的作用当队列产生新消息时服务端需要找到消费者对应的连接信息完成消息推送。b.消费者信息消费者标识 tag订阅队列名称队列产生消息时向该客户端推送客户端收到消息后针对该队列执行消息确认。自动确认标志自动确认消息推送完成后直接删除消息无需客户端回执确认。手动确认推送消息后等待客户端返回确认应答再删除消息。消费处理回调函数指针队列收到消息后调用该回调回调内部逻辑固定负责向对应客户端推送消息。c.消费者管理管理思想:以队列为单位管理消费者。消费者订阅、消息确认操作均绑定队列队列产生消息时取出该队列下的消费者信息来推送消息。队列消费者管理结构:数据信息消费者链表保存当前队列全部消费者。采用 RR 轮询策略每次选取下一个消费者推送消息一条消息只交付给一个客户端处理。管理操作新增消费者RR 轮询获取一个消费者删除消费者获取队列消费者数量判断消费者链表是否为空队列消费者结构生命周期操作初始化队列消费者结构删除队列消费者结构向指定队列添加消费者获取指定队列消费者删除指定队列消费者2.信道管理:Channel信道是网络通信里的通信通道概念。网络通信依靠网络连接完成为提升资源利用率会在通信连接之上进一步抽象出通信通道。 对使用者而言信道是网络通信载体一条底层网络连接可以创建多个信道。 各个信道在用户视角相互独立但底层共用同一条网络连接完成通信。信道作为面向用户的通信通道所有网络服务能力都由信道对外提供。a.信道提供的服务操作声明 / 删除交换机声明 / 删除队列绑定 / 解绑队列与交换机发布消息 / 订阅队列消息 / 取消队列订阅 / 队列消息确认b.信道需要管理的数据信道 ID信道关联的虚拟机句柄信道关联的消费者句柄信道关闭时需要取消该信道下全部消费者订阅等价于删除相关消费者工作线程池句柄信道完成消息投递到队列后从队列取出消费者执行消息消费把消息推送给客户端的任务交给线程池异步执行。并非每个信道单独持有线程池整个服务器共用一个线程池所有信道都复用这个线程池完成异步任务。c.信道的管理操作创建一个信道关闭一个信道获取指定信道句柄3.连接管理概念网络通信连接底层网络通信模块基于 muduo 库实现muduo 库原生提供 Connection 连接对象。 但 muduo 库本身不存在上层通信信道Channel的概念。 所以需要在应用层对 muduo 的 Connection 做二次封装实现自定义的连接管理。a.管理数据muduo 库的通信连接当前连接关联的信道管理句柄b.连接提供的操作创建信道关闭信道c.管理的操作新增连接关闭连接获取指定连接信息4. 异步工作池模块TcpClient 模块需要一个 EventLoopThread 模块负责 IO 事件监控客户端收到推送消息后需要对消息进行业务处理因此依靠线程池完成消息处理工作将异步工作线程模块独立封装的原因 多个连接可以共用同一个 EventLoopThread 来完成 IO 事件监控全部推送消息的处理共用同一个线程池即可。不需要为每一个连接单独创建 EventLoop也不需要为每个信道单独分配线程池。三.项目模块关系图四.总结消息队列系统的模块划分整体分为三大部分实现服务端发布客户端订阅客户端服务端模块1. 数据管理模块交换机数据管理模块队列数据管理模块绑定数据管理模块消息数据管理模块以上四个模块分别负责数据的增删查管理同时实现持久化存储。2. 虚拟机数据管理模块虚拟机本质是交换机 队列 绑定 消息组成的整体逻辑单元。 虚拟机的数据管理就是对上面四个模块做合并管理。3. 交换路由模块消息发布时消息投递到交换机由交换机决定消息分发到哪些队列。 交换机类型决定分发逻辑直连交换、广播交换、主题交换。 直连交换、广播交换逻辑简单主题交换需要做规则匹配交换路由模块就是负责该匹配流程。4. 消费者管理模块消费者指订阅某个队列消息的客户端队列产生消息后会向该客户端推送消息。 核心 API 提供队列订阅服务订阅的对象是队列不是单条消息。 当前主要实现消息推送能力队列有消息时可以查询到消费者及其对应的信道信息。5. 信道通信通道管理模块一条网络连接可以对应多个通信通道。 客户端关闭通信时关闭的不是底层连接而是自身对应的信道关闭信道时需要取消该客户端的订阅。6. 连接管理模块对应底层网络通信连接。 当连接关闭时需要关闭该连接关联的全部信道因此连接管理模块还需要维护关联信道信息。7. 服务端 BrokerServer 模块整合上面全部模块组装成完整消息队列服务。客户端模块1. 消费者管理模块订阅客户端订阅队列消息时等价于创建一个消费者。2. 信道管理模块客户端信道和服务端信道一一对应。服务端信道对外提供的能力客户端信道同样具备。 简单理解服务端给客户端提供服务客户端信道面向用户提供服务。3. 连接管理模块对用户而言所有业务操作都基于信道完成信道是用户视角的通信载体而非底层连接。 用户的所有请求都通过信道发起连接管理负责整合客户端侧资源。4. 基于以上三个模块封装实现两类客户端订阅客户端订阅指定队列消息接收推送消息并处理发布客户端向指定交换机发送消息