如何设计一款基于 MySQL 实现的 Message Queue

👉 这是一个或许对你有用的社群

🐱 一对一交流/面试小册/简历优化/求职解惑，欢迎加入「芋道快速开发平台」知识星球。下面是星球提供的部分资料：

• 《项目实战（视频）》：从书中学，往事上“练”
• 《互联网高频面试题》：面朝简历学习，春暖花开
• 《架构 x 系统设计》：摧枯拉朽，掌控面试高频场景题
• 《精进 Java 学习指南》：系统学习，互联网主流技术栈
• 《必读 Java 源码专栏》：知其然，知其所以然

• EMS
• 功能
• 写入设计
• 读取设计
• 核心表设计
• 锁🔐

EMS

基于 Spring Boot + MyBatis Plus + Vue & Element 实现的后台管理系统 + 用户小程序，支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能

• 项目地址：https://github.com/YunaiV/ruoyi-vue-pro
• 视频教程：https://doc.iocoder.cn/video/

功能

1. 集群消费、广播消费
2. 自动重试、死信队列
3. 快速重置消息位点，快速回放消息,快速查询消息
4. 消息可基于磁盘积压、消息可快速清理
5. 监控 group 积压，topic 消息量排行，消息链路追踪，消息消费超时告警；
6. 读写性能 1200-3000 QPS 左右

基于 Spring Cloud Alibaba + Gateway + Nacos + RocketMQ + Vue & Element 实现的后台管理系统 + 用户小程序，支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能

• 项目地址：https://github.com/YunaiV/yudao-cloud
• 视频教程：https://doc.iocoder.cn/video/

写入设计

• 针对一个 topic，需要有物理 physics offset， 每次写入，topic 维度的 physics_offset 自增加一
• 如果使用 redis 自增特性实现, 为每个 topic 配置一个自增 key, 则可以避免加锁.
• redis 实现虽然性能好, 如为配置aof,宕机则可能导致丢失数据, 此时,会出现 offset 重复异常, 过一会随着继续自增, 也就恢复了.
• 写入需要上锁吗? 看怎么写, 如果使用非原子的形式自增 id,比如数据的的方式,先查出最大 id,再加一,那么必须加锁
• topic 维度的自增 id 如果使用 mysql 实现， 性能不堪受辱，因此，此处使用 redis 自增实现（可配置为 mysql 实现）；
• 经过测试，笔记本电脑，单 topic 20 并发写入，qps 在 1000-1500 左右（local mysql & local redis），基本满足业务需求。
• 考虑到高可用性和业务场景，此处无法使用批量插入
• 所有的 topic 和 msg 都写入的这一张表中，表数据定时清理，消费完的消息，可提前删除。
• 注意，本方案写入性能瓶颈是 MySQL Server 的性能瓶颈。

读取设计

1. 假设针对一个 topic，只有一个 consumer，只需循环读取，然后更新 offset 即可。

• 但结合实际业务场景，这种基本不存在，所以，忽略这种场景。
• 通常，一个 topic 有多个 consumer group（简称 tg）， 一个 consumer group 有多个 client（jvm or thread）

2. 如果一个 topic + group（简称 tg），有多个 consumer，每个 consumer 有多个线程，读取和更新 offset 则会有并发问题, 如下图。

• 这个 client id，我们将其设计为，ip + pid + uuid + thread id；
• ip 和 pid 可帮助我们追溯问题
• uuid 简单防重复
• thread id，一种性能优化，下面细说。
• 结合实际业务场景，且遵循 simple is better 原则，读取时，使用上锁的方式解决并发问题。锁的粒度就是 tg
• 考虑到要实现基本的顺序读取和防止重复消费，多线程并发时，我们应当实现基于自增的形式读取 msg；每个 clientid 读取消息后，都会记录一个简单的log，并在 tg 维度增加一个 max offset
• 每次读取消息时，每个 client 都需要去检查当前想要读取的 tg 是否已经有【其他 client】在操作 max offset。即，我们将锁的粒度缩小到了 max offset；
• 对这个 tg 维度的 max offset + n
• 批量插入这个 tg + clientid offset log，表明这个消息被这个 clientid 读取了，同时也间接更新了 max offset（order by offset）
• 释放锁🔐
• 拉取刚刚读取的 msg id list 里面的消息体
• 交给业务处理消息
• 整体原则是，一个 t + g 的 max offset，同时只能有一个 thread 操作（写和更新）
• 如果有其他人在读取，则阻塞
• 如果没有其他人在读取，则锁住这个 tg， 并批量拉取一定数量的消息 id，

3. ack

• 对于集群消息，如何保证在断电情况下，消息不丢失,使用数据库存储消息, 写入即不会丢失, 但消费时, 如果刚刚读进内存就立刻宕机,则需要在重启时恢复消息.
• 每个 client get 到消息后，都需要记录 msg pid，consumer group，state（start、done，retry）为 start 状态
• ack success，将 log update 为 done 状态
• ack fail 后，将 log update 为 retry 状态，同时将消息存入重试队列
• 如果 client 还存活，超过 1 分钟(可配)，则将其捞出，放进重试队列，并在 10s 进行第一次重试
• 如果 client 还存活，则立刻将其捞出，放进重试队列，并在 10s 进行第一次重试
• 这里需要上锁吗？其实是不需要的，因为更新的维度是 client id 的 log，不存在并发更新. 这里更新状态是表示这些消息已经处理结束了，否则无法判定宕机场景。
• 对于 start 状态的消息，定时任务会去检查
• ack 是批量的，ack 失败，仅会导致重复消费。

4. 广播消息

• 是否为广播消息由 topic 确定
• 广播消息不需要上锁，每一个订阅该 topic 的 client 都会读取到该消息
• 广播消息不需要 ack，不需要记录成功或失败或重试，仅需要内存里记录 offset
• 推荐尽可能使用集群模式，使用集群模式模拟广播模式

5. client id

• 只有 consumer 需要 client id
• client id 由 ip pid uuid + thread id 组成, 可溯源.
• client id 需要续约（5s），如果机器宕机，则会被自动清除，且他的 start 状态的消息会进入重试队列，交给同 group 的其他 client
• client id 可以自己主动注销，注销前，自己内存的消息应当被优雅消费结束，一般来讲，kill -15 的 jvm 都会主动注销 client id；

核心表设计

1. topic 表：记录topic 元信息
2. group 表：记录 group 订阅元信息
3. msg 表：msg总表，记录写入的信息，包含 body 和 topic 维度的自增 offset，类似 rocketmq commit log

• 该表会被多个 consumer 消费的消息
• 该表会被定制删除过期数据

4. retry msg 表，消费失败、超时的消息，会进入该表，并按阶梯定时消费
5. dead msg 表，消费重试 16（any config） 次的消息，会进入该表
6. topic_group_log 表：记录 consumer group client 的 msg 消费记录，包含 state（start、done，retry） 字段，可 ack

• 该表的记录行数会非常多，单行数据较少，可自动删除 done 的记录

锁🔐