【仿 RabbitMQ 项目】发布订阅消息队列 (篇十)：《队列消息管理模块 —— 从逻辑分析到代码实现再到测试》---详解

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */
#content_views .toc,
/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */
#content_views.markdown_views > p:empty:has(+ .toc),
#content_views.markdown_views > .toc + p:empty,
/* 富文本旧版目录标记 */
#content_views.htmledit_views #main-toc,
#content_views.htmledit_views #hr-toc,
#content_views.htmledit_views p[id*=\”-toc\”] {
display: none !important;
}
/* 目录去掉后，紧跟的首个标题不再多出一块上边距 */
#content_views.markdown_views > .toc + h1,
#content_views.markdown_views > .toc + h2,
#content_views.markdown_views > .toc + h3,
#content_views.markdown_views > .toc + h4,
#content_views.markdown_views > .toc + p:empty + h1,
#content_views.markdown_views > .toc + p:empty + h2,
#content_views.markdown_views > .toc + p:empty + h3,
#content_views.markdown_views > .toc + p:empty + h4 {
margin-top: 0 !important;
}

这篇文章完整记录一个模块从需求 –> 设计 –> 编码 –> 测试的全过程：

• 第一部分：逻辑分析 —— 需求怎么拆成类，三个类为什么这么分
• 第二部分：代码实现 —— 三个类的完整实现 + 逐段解析
• 第三部分：类之间的关系 —— 调用链、数据流、职责边界
• 第四部分：测试验证 —— 真实踩过的坑 + 最终通过结果

所有报错信息和测试输出均为实际运行结果。

第一部分：逻辑分析

1.1 需求原文

先把需求原文完整贴出来 —— 后面的设计与代码，每一条都能对应回这里。

因为消息数据需要在网络中进行传输，因此消息的类型定义使用 protobuf 进行，因为 protobuf 中自带了序列化和反序列化功能，因此操作起来会简便一些。

需要特别说明的是，消息的存储并没有使用数据库，因为消息长度通常不定，且有些消息可能会非常庞大，因此并不适合存储在数据库中，因此我们的处理方式（包括 RabbitMQ）是直接将消息存储在文件中进行管理，而内存中管理的消息只需要记录好自己在文件中的所在位置和长度即可。

为了便于管理，消息管理以队列为单元进行管理，因此每个队列都会有自己独立的数据存储文件。

1. 创建消息类型的 proto 文件，并使用 protobuf 命令生成相对应的代码文件

• a. 属性：消息 ID、路由主题、持久化模式标志
• b. 消息内容
• c. 有效标志（持久化需要）
• d. 持久化位置（内存中）
• e. 持久化消息长度（内存中）

2. 消息的持久化管理

• a. 管理数据
  • i. 队列消息文件存储的路径
  • ii. 队列消息的存储文件名
  • iii. 队列消息的临时交换文件名
• b. 管理操作
  • i. 日志消息存储在文件中（4B 长度 + (属性+内容+有效位) 序列化消息，连续存储即可）
  • ii. 提供队列消息文件创建/删除功能
  • iii. 提供队列消息的新增持久化/删除持久化
  • iv. 提供持久化内容的垃圾回收（其实就是重新加载出所有有效消息返回，并重新生成新的消息存储文件）

3. 消息的管理（以队列为单位进行管理）

• a. 队列消息管理数据
  • i. 队列名称
  • ii. 待推送消息链表
  • iii. 持久化消息 hash
  • iv. 待确认消息 hash
  • v. 有效消息数量
  • vi. 已经持久化消息总量
  • vii. 持久化管理句柄
• b. 队列管理操作
  • i. 新增消息
  • ii. 获取队首消息（获取的同时将消息加入待确认队列）
  • iii. 移除指定待确认消息
  • iv. 获取队列待消费 & 待确认消息数量
  • v. 恢复队列历史消息
  • vi. 销毁队列所有消息
  • vii. 判断队列消息是否为空
• c. 消息的总体对外管理
  • i. 初始化新建队列的消息管理结构，并创建消息存储文件
  • ii. 删除队列的消息管理结构，以及消息存储文件
  • iii. 向指定队列新增消息
  • iv. 获取指定队列队首消息
  • v. 确认指定队列待确认消息（删除）
  • vi. 判断指定队列消息是否为空

需求中的三个核心决策

1.2 需求: 代码总映射表

需求里每一条，最终落在哪个类的哪个函数上。看懂这张表，整个模块的结构就清楚了。

需求 1：proto 消息定义

 d、e 两个字段的位置很关键：它们和 payload 同级（不是放在 payload 里面）， 所以 msg->payload().SerializeAsString() 不会把它们写进文件 —— 它们只是内存中的管理信息。 这也是为什么 load() 读出来之后必须重新计算并设置 offset / length。

需求 2：消息的持久化管理 —> MessageMapper

2b-i 的存储格式：

需求 3a：队列消息管理数据 → QueueMessage 的成员变量

需求要 7 项，代码里就是 7 个成员 + 1 个 _mutex。 "句柄"这个词在这里就是"一个成员对象"，不是指针 —— 因为它随 QueueMessage 一起创建、一起销毁。

需求 3b：队列管理操作 → QueueMessage 的成员函数

需求 3b-ii 那句"获取的同时将消息加入待确认队列"是重点：

MessagePtr front()
{
std::unique_lock<std::mutex> lock(_mutex);
if (_msgs.size() == 0) return MessagePtr();
MessagePtr msg = _msgs.front();
_msgs.pop_front();
_waitack_msgs.insert({ msg->payload().properties().id(), msg }); //<— 需求明确要求的这一步
return msg;
}

为什么要"移入待确认"而不是直接删？ 

因为消息投递出去后，消费者可能处理失败。只有收到 ack 才算真正消费完成 —— 这是 MQ 的可靠性保证。

需求 3c：总体对外管理 → MessageManager

注意 3c 只列了 6 项，但代码里还有 4 个统计函数（getable_count / total_count / durable_count / waitack_count）—— 它们是 3b-iv 在"对外层"的转发版本，需求没单列，但实现上必须有。

1.3 需求拆解出的三个类

这就是分层：每一层只关心自己那部分，下层的变化不影响上层。

为什么这么分?

如果全塞在一个类里会怎样？

class BadMessageManager {
// 既要管文件格式、又要管消息状态、又要管多队列查找
// 结果：2000 行，任何改动都牵一发动全身
};

分层之后：

1.4 存储格式设计

既然内存只记"位置和长度"，文件就必须能按顺序切分出每条消息

┌────────────┬──────────────────┬────────────┬──────────────────┐
│ 4B 长度 N │ payload(N 字节) │ 4B 长度 M │ payload(M 字节) |…
└────────────┴──────────────────┴────────────┴──────────────────┘
↑
msg->offset() 指向这里（长度字段之后）
msg->length() == N

为什么要存长度？ 

因为要能从文件里"顺序切分"出一条条消息，必须知道每条多长。

1.5 三个关键问题的权衡

问题一：删除消息怎么处理？

方案 A（物理删除）：把消息从文件里删掉，后面的数据整体前移。

删除第 1 条 –> 后面所有消息的位置全变 –> 需要重写整个文件 –> O(n)

方案 B（标记删除）：不删数据，只把有效标志改掉

消息结构里有 valid 字段：
"1" = 有效
"0" = 已作废
删除时：找到 offset，把 payload 覆盖写一遍，只改 valid 的值 –> O(1)

结论：选 B。 但会积累废数据，所以需要一个"整理"机制。

问题二：废数据怎么清理？

引入垃圾回收（GC）：当废数据占比过高时，把文件里所有有效消息重写到新文件。

触发条件：持久化总量 > 2000 且 有效比例 < 50%
执行方式：读有效消息 –> 写临时文件 –> 删旧文件 –> 临时文件改名

关键点：重写后每条消息的位置都变了，必须同步更新内存里的 offset / length。

问题三：valid 改动会不会破坏文件结构？

会，如果长度变了的话。 因为这是"原地覆盖写"：

"1" –> "0" 长度不变（都是 1 字节） 安全
"true" –> "false" 长度变了 会把下一条消息的数据覆盖掉

所以代码里必须有长度校验：

std::string body = msg->payload().SerializeAsString();
if (body.size() != msg->length()) {
DLOG("不能修改文件中的数据信息，因为新生成的数据与原数据长度不一致!");
return false;
}

1.6 一个队列需要管理什么

消息在队列里的状态是流动的，需要三个容器分别记录不同阶段的：

insert()
│
▼
┌────────────────────────┐
│ _msgs 待推送链表 │ 新来的、还没投递给消费者
└───────────┬────────────┘
│ front() 取出队首
▼
┌────────────────────────┐
│ _waitack_msgs 待确认 │ 已投递、等消费者 ack
└───────────┬────────────┘
│ ack -> remove(msg_id)
▼
真正删除

┌────────────────────────┐
│ _durable_msgs 持久化 │ 所有在文件里有记录的消息（按 id）
└────────────────────────┘
↑ insert 时加入，ack 时移除，recovery 时重建

为什么需要 _waitack_msgs？

 这是消息队列的可靠性保证：消息投递出去但还没被确认时，如果消费者挂了，消息不能丢 —— 必须记着它已经"发出去了"。

1.7 消息结构定义（protobuf）

// ============================================================================
// mq_msg.proto —— 消息队列的消息类型定义
// —————————————————————————-
// 生成 C++ 代码：
// protoc –cpp_out=. mq_msg.proto
// 会得到：
// mq_msg.pb.h （声明）
// mq_msg.pb.cc （实现，编译时必须一起链接！）
//
// 文件名建议统一（下面按 mq_msg.proto 命名）。
// 如果实际文件叫 message.proto，改一下 protoc 的输入即可，内容不变。
// ============================================================================
syntax = "proto3"; // proto3 语法：字段默认值 = 0 / "" / false，且不能自定义默认值
package bitmq; // 生成的类都在 bitmq 命名空间里
// —— 所以代码里写 bitmq::Message，而不是 ::Message
// ============================================================================
// 1. 交换机类型
// —————————————————————————-
// DIRECT : 精确匹配路由键 （news.sport → 只投给绑定 news.sport 的队列）
// FANOUT : 广播，忽略路由键 （投给所有绑定的队列）
// TOPIC : 按通配符匹配 （news.* / news.# / *.sport）
// ============================================================================
enum ExchangeType {
UNKNOWTYPE = 0; // ★ proto3 枚举必须有一个 0 值，这里用它表示"未设置/未知"
DIRECT = 1; // 直接交换机
FANOUT = 2; // 广播交换机
TOPIC = 3; // 主题交换机
};
// ============================================================================
// 2. 持久化模式（消息属性之一）
// —————————————————————————-
// UNDURABLE : 不持久化 —— 只放内存，进程重启就丢
// DURABLE : 持久化 —— 写进 .mqd 文件，重启后能恢复
//
// 注意 UNKNOWMODE = 0：
// proto3 里"没显式设置的枚举字段"就是 0，
// 所以只要没调用 set_delivery_mode()，拿到的就是 UNKNOWMODE（= 0）。
// 代码里判断是否落盘时，必须显式比较 == DeliveryMode::DURABLE，
// 绝不能写成 != UNDURABLE —— 那样 UNKNOWMODE 也会被当成持久化。
// ============================================================================
enum DeliveryMode {
UNKNOWMODE = 0; // 未设置（默认值）
UNDURABLE = 1; // 非持久化
DURABLE = 2; // 持久化
};
// ============================================================================
// 3. 消息属性（需求 1a）
// —————————————————————————-
// 它相当于消息的"信封/快递单"：描述这条消息是什么、该往哪投、要不要存盘。
// 真正的业务数据放在 Message.Payload.body 里，两者分开是为了：
// · 路由/投递只需要读属性，不用碰可能很大的 body
// · body 是任意字符串，属性是结构化字段
// ============================================================================
message BasicProperties {
string id = 1; // 消息唯一 ID（uuid 生成）
DeliveryMode delivery_mode = 2; // 持久化模式（决定要不要落盘）
string routing_key = 3; // 路由主题（交换机按它决定投给哪些队列）
};
// ============================================================================
// 4. 消息本体（需求 1b ~ 1e）
// —————————————————————————-
// Payload —— 会被【序列化写进文件 / 走网络传输】的部分
// offset —— 【只在内存中】，不参与序列化
// length —— 【只在内存中】，不参与序列化
//
// 这个层级关系是理解整个存储模块的关键 ★★
//
// Message {
// payload ← 序列化的是它（调用 msg->payload().SerializeAsString()）
// offset ← 和 payload 同级！不会进文件
// length ← 同上
// }
//
// 所以从文件 load() 出来之后，offset / length 全是 0，
// 必须由 load() 自己重新计算并 set 回去：
//
// msgp->set_offset(offset – msg_size);
// msgp->set_length(msg_size);
//
// 少了这两行，之后再 ack（原址覆盖写）就会从文件开头开始写 —— 直接毁掉文件。
// ============================================================================
message Message {
// 内嵌类型：序列化时用的是它
message Payload {
BasicProperties properties = 1; // 属性（ID / 持久化模式 / 路由键）
string body = 2; // 消息内容（真正的业务数据）
string valid = 3; // 有效标志："1" = 有效，"0" = 已作废
// （标记删除用，必须保持定长！）
};
Payload payload = 1; // 序列化对象：写文件、网络传输的都是它
uint32 offset = 2; // 数据体在 .mqd 文件中的起始位置（内存中，不落盘）
uint32 length = 3; // 数据体的字节数（内存中，不落盘）
};

注意 offset / length 和 payload 是同级字段：

msg->payload().SerializeAsString() // <— 只序列化 payload，offset/length 不进文件
msg->set_offset(…) // <— 这是内存中的管理信息
msg->set_length(…)

第二部分：代码实现

说明：以下代码基于实际项目整理，关键处与真实运行结果对应。

2.1 公共头文件与类型别名

MessagePtr 展开后：

MessagePtr msg = std::make_shared<Message>();
// 等价于
std::shared_ptr<bitmq::Message> msg = std::make_shared<bitmq::Message>();

2.2 第一层：MessageMapper（文件持久化）

完整实现

关键点解析:

① 为什么 offset 要 + sizeof(size_t)？

写入位置 fsize：从这里开始写 4 字节长度
↓
┌────────────┬──────────────────┐
│ 长度(4B) │ payload │
└────────────┴──────────────────┘
fsize fsize + 4 ← offset 指向这里

所以 msg->set_offset(fsize + sizeof(size_t))。

② load() 里为什么 set_offset(offset – msg_size)？

循环里 offset 是移动过后的值：

读长度前：offset = 本条长度字段的起点
读长度后：offset += 4
读数据后：offset += msg_size ← 此时 offset 指向"下一条的起点"

所以本条 payload 的起点 = offset – msg_size

这两处 set_offset 缺一不可 —— 缺了 load() 里的，恢复出来的消息 offset 全是 0，之后 ack 会从文件头开始覆盖写，直接把文件写坏。

③ GC 为什么用"写新 → 删旧 → 改名"

写 .tmp → 删 .mqd → .tmp 改名成 .mqd

任何一步崩溃，原文件都还在（直到第 2 步才删）。如果直接原地重写，中途失败就两头都没有了。

2.3 第二层：QueueMessage（单队列消息管理）

完整实现

关键点解析

① _valid_count vs _total_count

size_t _valid_count; // 文件里"有效"的消息数（valid=="1"）
size_t _total_count; // 文件里"占用过"的记录数（含已作废的）

举例：插 10 条持久化消息，ack 掉 3 条

GC 触发条件就是基于这两个值：

if (_total_count > 2000 && _valid_count * 10 / _total_count < 5)
// ↑ 用整数运算等价于"有效比例 < 50%"，避免浮点

② gc() 为什么不能自己加锁

bool remove(const std::string &msg_id) {
std::unique_lock<std::mutex> lock(_mutex); // 已持锁
…
gc(); // ← 这里调用
}

std::mutex 不可重入，gc() 里再加锁就死锁。所以设计成"调用者必须已持锁"的私有函数。

③ insert 里"是否落盘"的判断依据

// 正确：看消息自己的 delivery_mode
if (msg->payload().properties().delivery_mode() == DeliveryMode::DURABLE)

不能用"队列是否持久化"去判断 —— 那会导致 UNDURABLE 的消息也被写进文件（这就是测试中踩到的坑 3）。

"持久化"是消息级别的属性。

2.4 第三层：MessageManager（对外总管理）

完整实现

关键点解析

① 统一套路："锁内查表、锁外干活"

QueueMessage::ptr qmp;
{
std::unique_lock<std::mutex> lock(_mutex); // 只保护 map
auto it = _queue_msgs.find(qname);
if (it == _queue_msgs.end()) { DLOG(…); return …; }
qmp = it->second; // 拷一份 shared_ptr
} // 锁在这里释放
return qmp->insert(…); // 真正的活在锁外做

为什么？

拷贝 shared_ptr 也是必要的 —— 保证锁释放后对象不会被别的线程 destroyQueueMessage 析构掉。

② 为什么 initQueueMessage 里 recovery() 在锁外

{
lock;
qmp = make_shared<QueueMessage>(…);
_queue_msgs.insert({qname, qmp});
}
qmp->recovery(); // ← 锁外，因为要读文件

第三部分：三个类之间的关系

3.1 静态关系

┌─────────────────────────────────────────────────────────┐
│ MessageManager │
│ ┌───────────────────────────────────────────────────┐ │
│ │ _queue_msgs: unordered_map<string, QueueMessage::ptr> │
│ │ "queue1" ──► QueueMessage 对象 A │ │
│ │ "queue2" ──► QueueMessage 对象 B │ │
│ └───────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘
│ 每个 QueueMessage 内部持有
▼
┌─────────────────────────────────────────────────────────┐
│ QueueMessage │
│ ┌───────────────────────────────────────────────────┐ │
│ │ _msgs list<MessagePtr> 待推送 │ │
│ │ _waitack_msgs map<id, MessagePtr> 待确认 │ │
│ │ _durable_msgs map<id, MessagePtr> 持久化 │ │
│ └───────────────────────────────────────────────────┘ │
│ _mapper: MessageMapper ◄── 组合关系（一个队列一个） │
└─────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ MessageMapper │
│ _datafile: ./data/queue1.mqd │
│ _tmpfile: ./data/queue1.mqd.tmp │
│ （只有字符串路径，不持有 Message 对象） │
└─────────────────────────────────────────────────────────┘

3.2 关系类型

为什么用 shared_ptr？

同一条消息会被多个容器同时持有：

_msgs.push_back(msg); // 待推送链表持有
_durable_msgs.insert({id, msg}); // 持久化 hash 又持有
// front() 之后：
_waitack_msgs.insert({id, msg}); // 待确认 hash 再持有

用裸指针就得手动管理生命周期，删早了其他容器变悬空指针。shared_ptr 用引用计数自动解决。

3.3 调用链（一次完整的消息流转）

insert —— 从上层到下层的调用链

应用层：mmp->insert("queue1", &props, "Hello World-1", DeliveryMode::DURABLE)
│
▼
MessageManager::insert
├─ 锁 _mutex，在 _queue_msgs 里找 "queue1" → 拿到 qmp
├─ 释放锁
└─ qmp->insert(bp, body, mode)
│
▼
QueueMessage::insert
├─ 构造 Message 对象，设置 payload.body / properties
├─ 锁 _mutex
├─ delivery_mode == DURABLE？
│ 是 → _mapper.insert(msg)
│ │
│ ▼
│ MessageMapper::insert
│ ├─ 序列化 payload
│ ├─ 写 4 字节长度到文件末尾
│ ├─ 写数据体
│ └─ msg->set_offset / set_length
│ 然后 _valid_count++、_total_count++、_durable_msgs.insert
└─ _msgs.push_back(msg)

front —— 消息状态的迁移

MessageManager::front("queue1")
└─ QueueMessage::front
├─ _msgs.front() 取出队首
├─ _msgs.pop_front() ← 从待推送移除
├─ _waitack_msgs.insert({id, msg}) ← 加入待确认
└─ return msg

注意：front() 不碰文件。文件只在 insert（追加）和 ack（标废）时改动。

ack —— 标记删除的完整路径

MessageManager::ack("queue1", msg_id)
└─ QueueMessage::remove(msg_id)
├─ 在 _waitack_msgs 找 msg_id
├─ delivery_mode == DURABLE？
│ 是 → _mapper.remove(msg)
│ │
│ ▼
│ MessageMapper::remove
│ ├─ set_valid("0")
│ ├─ 序列化
│ ├─ ★ 长度校验（必须与长度字段一致）
│ └─ 从 msg->offset() 处覆盖写
│ _durable_msgs.erase(msg_id)
│ _valid_count -= 1
│ gc() ← 判断是否需要整理文件
└─ _waitack_msgs.erase(msg_id)

recovery —— 从磁盘重建内存

MessageManager::initQueueMessage("queue1")
├─ 锁内：new QueueMessage → 构造 MessageMapper（建目录/建文件）
├─ 锁内：放进 _queue_msgs
└─ 锁外：qmp->recovery()
└─ QueueMessage::recovery
├─ _msgs = _mapper.gc()
│ │
│ ▼
│ MessageMapper::gc
│ ├─ load() 扫描文件，跳过 valid=="0"
│ ├─ 逐条写入 .tmp（同时刷新 offset/length）
│ ├─ 删掉旧 .mqd
│ └─ .tmp 改名为 .mqd
├─ 填充 _durable_msgs
└─ _valid_count = _total_count = 消息数

3.4 数据流全景图

┌────────────────────────────────────────────┐
│ 应用 / 网络层 │
└────────────────────────────────────────────┘
│insert │front │ack
▼ ▼ ▼
┌────────────────────────────────────────────┐
│ MessageManager │
│ 锁内查表 → 拿 qmp → 锁外调用 │
└────────────────────────────────────────────┘
│
▼
┌────────────────────────────────────────────┐
│ QueueMessage │
│ │
│ _msgs ──front()──► _waitack_msgs │
│ │ │ │
│ │insert │ack │
│ ▼ ▼ │
│ _durable_msgs ◄──── 标记删除 │
└────────────────────────────────────────────┘
│ ▲
│ 写 │ 恢复
▼ │
┌────────────────────────────────────────────┐
│ MessageMapper │
│ .mqd 文件（4B长度 + payload 连续排列） │
└────────────────────────────────────────────┘

3.5 职责边界表

判断标准：如果一个改动需要动两层，说明职责划错了。

第四部分：测试验证

4.1 测试用例设计

核心难点：持久化模块必须"分两轮"测

4.2 实战踩坑记录

坑 1：undefined reference to bitmq::Message::Message()

现象：

原因：undefined reference 是链接错误。bitmq::Message 是 protobuf 从 .proto 生成的类，实现都在 mq_msg.pb.cc 里，而 Makefile 只编了测试源文件。

解决：

.pb.cc 必须一起编译；标准要 C++17。

坑 2：cannot convert 'bool' to 'bitmq::DeliveryMode'

现象：

逐句读这条报错：

原因：QueueMessage::insert 的第 3 个参数已改成 DeliveryMode，但 MessageManager::insert 还是 bool。

解决：全链路一起改。

bool insert(const std::string &qname, BasicProperties *bp,
const std::string &body, DeliveryMode mode) // bool → DeliveryMode
{
// …
return qmp->insert(bp, body, mode); // 转发也用 mode
}

坑 3：total_count 期望 4，实际 5

现象：

原因：QueueMessage::insert 里判断是否落盘用错了依据 —— 用了"队列是否持久化"去覆盖"消息自己的 delivery_mode"，导致第 5 条 UNDURABLE 的消息也被写进了文件。

解决：看消息自己的 delivery_mode。

// 设置属性时用调用方传的 mode
msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);

// 判断落盘时看"消息自己的 mode"
if (msg->payload().properties().delivery_mode() == DeliveryMode::DURABLE) {
msg->mutable_payload()->set_valid("1");
_mapper.insert(msg);
_valid_count += 1;
_total_count += 1;
_durable_msgs.insert({ msg->payload().properties().id(), msg });
}
_msgs.push_back(msg); // 不管持不持久化，都进待推送

教训："持久化"是消息级别的属性，不是队列级别的。这个 bug 很隐蔽：getable_count == 5 是对的，只有持久化相关计数多 1。

坑 4：getable_count 期望 5，实际 10（数字翻倍）

现象：

原因：跑测试前没删数据目录。

SetUp() → initQueueMessage() → recovery() 把上次的数据恢复了，于是 5 旧 + 5 新 = 10。

解决：

rm -rf data/ && ./mq_messagetest

教训："恢复"做对了，反而会让没清数据的测试出错。 排查技巧：看测试开始前有没有日志输出 —— 有日志 = 有恢复数据 = 数据没清干净。

坑 6：recovery_test 断言写错（第一条不是 World-1）

现象：

原因：测试的期望值错了，不是代码错了。

恢复日志证明实现是对的：

[DBG] 加载到无效消息: Hello World-1 ← 正确跳过了被 ack 作废的
[DBG] 向临时文件写入数据: Hello World-2
[DBG] 向临时文件写入数据: Hello World-3
[DBG] 向临时文件写入数据: Hello World-4 ← 恢复了 3 条

时间线：

第一轮 insert_test → 写入 4 条持久化：World-1, 2, 3, 4（World-5 是 UNDURABLE）
第一轮 delete_test → ack(World-1) → 文件里 valid 改成 "0"

磁盘最终状态：
① World-1 valid="0" ← 作废
② World-2 valid="1"
③ World-3 valid="1"
④ World-4 valid="1"

第二轮 recovery() → 恢复 3 条：World-2, 3, 4

解决：修正断言。

注意：原测试在第一次 front() 之后没有判空就直接 msg1->payload() 了。 一旦 front() 返回空指针，这里就是空指针解引用（崩溃）。 加上 ASSERT_NE(msg1.get(), nullptr) 之后，即使取不到消息也只会断言失败，不会段错误。

教训：被 ack 的消息，重启后不该复活 —— 这正是"标记删除"的意义。 断言失败时先看日志：日志显示实现正确 → 问题在期望值，别急着改代码。

4.3 最终通过的结果

第 1 轮：插入 + 删除

打开第一轮：

打开插入，新增和删除测试：

运行结果：

第 2 轮：从磁盘恢复

关闭第一轮，打开第二轮测试:

第 3 轮：销毁

依旧保持第二轮开启，关闭第一轮：

5 个用例汇总

4.4 测试验证了哪些机制

最值得记住的验证点

delete_test 里这两个数字的差异，正是"标记删除"实现的证明：

ASSERT_EQ(mmp->durable_count("queue1"), 3); // 内存里持久化 hash 减 1
ASSERT_EQ(mmp->total_count("queue1"), 4); // 文件里记录数不变（只标 valid="0"）

删除前：文件 4 条记录（全 valid="1"）
ack 一条后：
durable_count = 3 ← 有效消息少了一个
total_count = 4 ← 文件里的记录还在，只是被标成作废

如果这两个数字都变成 3，说明实现成了"物理删除" —— 那就失去了"O(1) 删除 + 定期 GC 整理"的性能优势。