项目二：从零敲出一个 LSM-Tree KV 存储引擎（实战动手篇）
项目二实战动手篇：从零敲出一个 LSM-Tree KV 存储引擎
第〇部分：动手前的装备清单
笔记写给谁看：从没敲过一行代码的大一同学。上一篇概念篇你已经读过了，知道 WAL、MemTable、SSTable、布隆过滤器是什么。但知道和写出来是两回事——这一篇的目标只有一个：你亲手把一个 600 行的 LSM 引擎敲出来、编译通过、亲眼看到它跑起来。敲完它，你简历上"自开发 LSM-Tree 存储引擎及 Redis 协议服务器"这一行，每一个词你都能自己解释。
写代码和写作文不一样，作文有纸就行，代码需要四样东西。
第一样叫编辑器，就是你打字的地方。我们用 VS Code，去官网下载安装，一路下一步。第二样叫编译器，它是把人写的代码翻译成机器能跑的程序的工具。C++ 的代码机器看不懂，要靠编译器翻译。我们用 g++，在 Ubuntu 虚拟机里打开终端（就是那个黑框框，按 Ctrl+Alt+T 呼出来），敲 sudo apt install build-essential cmake，装完敲 g++ --version，看到版本号就是装好了。如果你在 Windows 上用 MSYS2，命令一模一样。第三样叫终端，就是黑框框本身，编译、运行程序都在这里敲命令。第四样叫文件夹（专业点叫目录），我们给这个项目建一个文件夹，所有代码都放里面。终端里敲 mkdir lsm_lab && cd lsm_lab，意思是"建一个叫 lsm_lab 的文件夹然后进去"。
每一课的节奏固定五步：新建文件敲代码、编译、运行、看输出、做练习。代码必须亲手敲，不许复制粘贴——手指的记忆比眼睛可靠，而且敲错、改错、修好，这个过程才是真正学到东西的地方。
每个里程碑我都标注了对应的简历词条，敲完你就能对上号。
M0第一课 M0：人生第一行代码
目标 屏幕上打印出一行字。就这么简单，但你要亲手走完"写代码 → 编译 → 运行"的完整流程。
在终端里敲：
mkdir m0_hello && cd m0_hello然后打开 VS Code，打开这个文件夹，新建文件 hello.cpp，敲下面五行（注意都是英文标点，分号不能少）：
#include <iostream>
int main() {
std::cout << "Hello, 存储! 我马上要开始写数据库了" << std::endl;
return 0;
}逐行讲。第一行 #include <iostream> 是"我要用输入输出工具"，iostream 就是 input/output stream（输入输出流）的缩写，std::cout 这个打印工具就住在里面。少了这行，编译器不认识 cout 是什么。第三行 int main() { ... } 是程序的入口，int 表示这个函数跑完会交回一个整数，每个 C++ 程序都从这里开始执行，花括号 {} 圈住它的身体。第四行 std::cout << "..." 把引号里的字符串"推"到屏幕上，<< 你就理解成"把右边的东西塞到左边的水管里"，std::endl 是"换行"。第五行 return 0 向操作系统报告"我正常跑完了"，0 就是"一切正常"的暗号，出错的程序习惯返回别的数字。
保存文件（Ctrl+S），回终端敲：
g++ hello.cpp -o hello
./hello第一条命令是编译：g++ 是编译器，hello.cpp 是源代码，-o hello 的意思是"输出的程序叫 hello"。编译没报错的话，文件夹里会多出一个叫 hello（Windows 上是 hello.exe）的可执行文件——这才是机器真正能跑的东西。第二条命令 ./hello 运行它，./ 的意思是"当前文件夹里的"。
你应该看到：
Hello, 存储! 我马上要开始写数据库了调试 。现在我们故意制造三个错误，让你见识报错长什么样。第一，删掉第四行末尾的分号再编译，编译器会说 expected ';'（我本来以为这里有个分号）——C++ 每条语句必须用分号结尾，这是它的语法。第二，把英文引号敲成中文引号 ""，报错会是一堆乱码或者 stray 开头的提示——写代码永远用英文标点，这是新手第一大坑。第三，把 main 拼成 mian，报错说 undefined reference to main——编译器找不到入口。看懂报错是程序员的核心技能：从第一行报错读起，它通常告诉你哪个文件哪一行出了什么问题。
练习 把打印的内容改成你自己的名字，再打印第二行，试试连续两个 << 接力：std::cout << "a" << "b" << std::endl;。
M1第二课 M1：最朴素的 KV 存储有多慢（全文之魂）
目标 一个用纯文本文件当存储的 KV 库。key 是 "user:123"，value 是 "video:45"，就像抖音的点赞记录。然后你亲眼看到它的 get 慢成什么样。
新建文件夹 m1_naive_log，新建 naive_kv.cpp：
#include <iostream>
#include <fstream>
#include <string>
#include <chrono>
int main() {
// ============ 第一部分：put 10000 条（追加写）============
std::ofstream out("data.log", std::ios::app);
auto t0 = std::chrono::steady_clock::now();
for (int i = 0; i < 10000; i++) {
out << "user:" << i << " video:" << (i % 100) << "\n";
}
out.flush();
out.close();
auto t1 = std::chrono::steady_clock::now();
long long write_ms =
std::chrono::duration_cast<std::chrono::milliseconds>(t1 - t0).count();
std::cout << "put 10000 条耗时: " << write_ms << " 毫秒\n";
// ============ 第二部分：get 1000 次（每次从文件头扫到尾）============
long long total_lines = 0;
int found = 0;
auto t2 = std::chrono::steady_clock::now();
for (int q = 0; q < 1000; q++) {
std::string target = "user:" + std::to_string(9000 + (q % 1000));
std::ifstream in("data.log"); // 每次查询都重新打开文件，从头读
std::string k, v;
while (in >> k >> v) { // 一行一行扫
total_lines++;
if (k == target) {
found++;
break; // 找到了，但是已经扫了这么多行
}
}
}
auto t3 = std::chrono::steady_clock::now();
long long read_ms =
std::chrono::duration_cast<std::chrono::milliseconds>(t3 - t2).count();
std::cout << "get 1000 次耗时: " << read_ms << " 毫秒\n";
std::cout << "平均每次 get 扫描行数: " << total_lines / 1000 << " 行\n";
std::cout << "找到次数: " << found << "/1000\n";
return 0;
}新东西逐个讲。#include <fstream> 引入文件工具，ofstream 是"往文件写的流"（o = output），ifstream 是"从文件读的流"（i = input）。std::ios::app 里的 app 是 append（追加），意思是"每次写都接在文件末尾，不动前面的内容"——这就是顺序写，磁盘最擅长的姿势。auto t0 = std::chrono::steady_clock::now() 是掐秒表，auto 的意思是"让编译器自己推断这个变量是什么类型"，chrono 是 C++ 的计时工具库。for (int i = 0; i < 10000; i++) 是循环：先建一个从 0 开始的计数器 i，每次循环完 i 加 1，i 到 10000 就停。std::to_string(i) 把数字变成字符串，"user:" + std::to_string(i) 把两段字符串拼起来。while (in >> k >> v) 是读取循环：in >> k 从文件里读一个词放进 k，读到词就继续，读不到（文件完了）就停。break 是"跳出循环"。
编译运行：
g++ -std=c++17 -O2 naive_kv.cpp -o naive_kv
./naive_kv-std=c++17 指定用 C++17 标准（我们全程用它），-O2 是开优化。你应该看到类似：
put 10000 条耗时: 3 毫秒
get 1000 次耗时: 1966 毫秒
平均每次 get 扫描行数: 19500 行
找到次数: 1000/1000写 10000 条只要 3 毫秒，读 1000 次要将近 2 秒，平均每次查询要扫 19500 行。这就是概念篇说的"翻完一百万个箱子"。put 为什么快？因为只做顺序追加，写完就不管了。get 为什么慢？因为 key 都堆在文件里没有任何组织，找一个 key 只能从头一个个对。
记住这个体感：我们后面每一课的每一行代码，都是在消灭这次全文件扫描。
调试 in >> k >> v 要求文件里每行恰好是"key 空格 value"两段，如果你的文件里有奇数个词，对不齐就会错位——数据格式一定要整齐。另外第一次运行完文件夹里会出现 data.log，重复实验前先 rm data.log，否则旧数据会混进新实验。
练习 把查询目标改成 0 到 999（文件开头的 key），看平均扫描行数变成多少。想想为什么差这么多——答案是命中越早 break 得越早。
M2第三课 M2：WAL + MemTable，断电也不怕
目标 put 直接查内存（快），同时写一份日志（安全）。然后你亲手"杀"掉程序再重启，数据一条不丢。
概念篇讲过：WAL（Write-Ahead Log，写前日志）就是记账的流水本，MemTable 就是内存里的有序表。现在把这套变成代码。新建 m2_wal_memtable 文件夹和 mini_kv.cpp：
#include <iostream>
#include <fstream>
#include <map>
#include <string>
// MiniKV：WAL + MemTable 版本
class MiniKV {
public:
MiniKV() {
// 第一步：启动时先看 wal.log 里有没有上次没来得及落盘的数据
std::ifstream in("wal.log");
std::string k, v;
while (in >> k >> v) {
memtable_[k] = v; // 重放（replay）：照着日志再做一遍
}
std::cout << "[启动] 从 wal.log 恢复了 " << memtable_.size() << " 条\n";
// 第二步：以"追加"方式打开 wal.log，以后的每次 put 都先记在这里
wal_.open("wal.log", std::ios::app);
}
void put(const std::string& key, const std::string& value) {
wal_ << key << " " << value << "\n"; // 顺序一：先写日志（落盘）
wal_.flush(); // flush 是把缓冲区真正推到磁盘
memtable_[key] = value; // 顺序二：再写内存
}
bool get(const std::string& key, std::string& value) {
std::map<std::string, std::string>::iterator it = memtable_.find(key);
if (it == memtable_.end()) return false; // 没找到
value = it->second; // 找到了，把值带出去
return true;
}
private:
std::map<std::string, std::string> memtable_; // 内存里的有序表
std::ofstream wal_; // 写前日志文件句柄
};
int main(int argc, char* argv[]) {
MiniKV db; // 一构造，就自动恢复了上次的数据
if (argc >= 2 && std::string(argv[1]) == "write") {
for (int i = 1; i <= 50; i++) {
db.put("msg:" + std::to_string(i), "dialog_" + std::to_string(i));
}
std::cout << "[写入] 50 条对话已保存\n";
} else if (argc >= 2 && std::string(argv[1]) == "read") {
int count = 0;
for (int i = 1; i <= 50; i++) {
std::string v;
if (db.get("msg:" + std::to_string(i), v)) count++;
}
std::cout << "[读取] 50 条里找到了 " << count << " 条\n";
} else {
std::cout << "用法:\n ./mini_kv write 写入 50 条\n ./mini_kv read 读取并验证\n";
}
return 0;
}这一课的重头戏是类（class）。class MiniKV { ... }; 把数据（memtable_、wal_）和操作数据的函数（put、get）捆在一起，外面的人只管调 db.put(...)，不用关心里面怎么实现的——这叫封装。public: 下面的东西是"外界可以用的"，private: 下面的是"只许类自己碰的"，别人碰不了，防止乱改。MiniKV() 这种"和类同名、没有返回值"的函数叫构造函数，对象一出生就自动执行，我们把"恢复数据"放在这里，保证谁用这个类都忘不了恢复。
新语法逐个过。std::map<std::string, std::string> 是一棵内存里的有序表（底层是红黑树，你暂时就当"自动排序的字典"），尖括号里写明 key 和 value 各是什么类型。memtable_[k] = v 存或覆盖一个键值对。memtable_.find(key) 找 key，返回一个迭代器（iterator），你可以理解成"指向表中某一格的游标"；找不到时它等于 memtable_.end()（一个"表尾之外"的特殊标记）。it->second 取出这格的 value，-> 意思是"游标所指的东西的成员"。char* argv[] 和 int argc 是命令行参数：你敲 ./mini_kv write 时，argc 是 2，argv[1] 是 "write"，程序由此知道该干哪个活。
put 里的顺序是灵魂：先写日志并 flush，再动内存。为什么这个顺序？假设反过来，先写内存再写日志，如果在两步之间断电，内存没了、日志也没记上，数据就真丢了。先落盘日志，就算内存的后续动作没做成，重启时照着日志重放一遍，数据还是回来了。
编译运行，做一次完整的"断电实验"：
g++ -std=c++17 -O2 mini_kv.cpp -o mini_kv
rm -f wal.log
./mini_kv write
./mini_kv read你应该看到：
[启动] 从 wal.log 恢复了 0 条
[写入] 50 条对话已保存
[启动] 从 wal.log 恢复了 50 条
[读取] 50 条里找到了 50 条第二次运行时，上一个进程早就不存在了，但数据一条不少——它们躺在 wal.log 里，新进程启动时重放恢复。现在更狠一点：write 之后不等它自己结束，按 Ctrl+C 直接杀掉进程，再 read，结果一样。这就是简历上"WAL 崩溃恢复"四个字的全部内容。
调试 忘写 wal_.flush() 的话，断电实验会丢最后几条——日志内容还在缓冲区没到磁盘，这就是 flush 的意义。另外 ./mini_kv 不带参数会打印用法，argc 这个名字是 argument count（参数个数）的缩写。
练习 给 MiniKV 加一个 count() 函数，返回 memtable_.size()，在 read 分支里打印总条数。
M3第四课 M3：手写跳表，把"有序"变成"快"
目标 一个 130 行的跳表（Skip List）。乱序插入 10 个学号，它能按序遍历、快速查找，你还能打印出"先坐快车再坐慢车"的查找路线。
为什么 MemTable 要有序？因为满了以后要按序落盘成 SSTable（下一课）。为什么不用 std::map 就好？——真实引擎（LevelDB/RocksDB）的 MemTable 是跳表，因为跳表并发更友好、实现可控。面试问 MemTable，答跳表是加分项。新建 m3_skiplist/skiplist.cpp：
#include <iostream>
#include <string>
#include <vector>
#include <cstdlib>
const int MAX_LEVEL = 8; // 快车道最多修 8 层
// 链表的一个结点
struct Node {
std::string key;
std::string value;
std::vector<Node*> forward; // forward[i] = 我在第 i 层的"下一站"
Node(const std::string& k, const std::string& v, int level)
: key(k), value(v), forward(level, nullptr) {}
};
// 跳表：有序 + 多层快车道
class SkipList {
public:
SkipList()
: head_(new Node("", "", MAX_LEVEL)), level_(1), size_(0) {
srand(42); // 固定随机种子，每次运行结果一样，方便你对照
}
~SkipList() {
Node* p = head_->forward[0];
while (p) {
Node* nxt = p->forward[0];
delete p;
p = nxt;
}
delete head_;
}
int size() const { return size_; }
// 插入或更新。返回 true 表示新插入，false 表示覆盖旧值
bool put(const std::string& k, const std::string& v) {
// update[i] 记住：每一层里，新结点要接在谁的后面
std::vector<Node*> update(MAX_LEVEL, head_);
Node* cur = head_;
// 从最高层往下找插入位置（和 get 的走法一样）
for (int lv = level_ - 1; lv >= 0; lv--) {
while (cur->forward[lv] && cur->forward[lv]->key < k) {
cur = cur->forward[lv];
}
update[lv] = cur;
}
Node* next = cur->forward[0];
if (next && next->key == k) { // key 已存在，覆盖
next->value = v;
return false;
}
// 抛硬币决定这个结点修几层
int lv = random_level();
if (lv > level_) { // 比现有楼还高，把头结点的接线也更新
for (int i = level_; i < lv; i++) update[i] = head_;
level_ = lv;
}
Node* node = new Node(k, v, lv);
// 逐层"接线"：新结点插到 update[i] 和它原来下一站之间
for (int i = 0; i < lv; i++) {
node->forward[i] = update[i]->forward[i];
update[i]->forward[i] = node;
}
size_++;
return true;
}
bool get(const std::string& k, std::string& v) {
Node* cur = head_;
for (int lv = level_ - 1; lv >= 0; lv--) {
while (cur->forward[lv] && cur->forward[lv]->key < k) {
cur = cur->forward[lv]; // 这一层能跳就跳
}
}
Node* next = cur->forward[0];
if (next && next->key == k) {
v = next->value;
return true;
}
return false;
}
// 教学用：打印查找 "88" 的完整路线（先坐快车再坐慢车）
void trace_get(const std::string& k) {
std::cout << "查找 " << k << " 的路线:\n";
Node* cur = head_;
for (int lv = level_ - 1; lv >= 0; lv--) {
while (cur->forward[lv] && cur->forward[lv]->key < k) {
std::cout << " 第 " << lv << " 层跳到 " << cur->forward[lv]->key << "\n";
cur = cur->forward[lv];
}
}
Node* next = cur->forward[0];
if (next && next->key == k)
std::cout << " 第 0 层命中: " << next->key << " = " << next->value << "\n";
else
std::cout << " 没找到\n";
}
// 教学用：打印每一层长什么样
void print_levels() {
for (int lv = level_ - 1; lv >= 0; lv--) {
std::cout << "第 " << lv << " 层: ";
for (Node* p = head_->forward[lv]; p; p = p->forward[lv]) {
std::cout << p->key << " ";
}
std::cout << "\n";
}
}
// 教学用：沿第 0 层走一遍（就是排序结果）
void print_all() {
std::cout << "第 0 层从头到尾（已有序）: ";
for (Node* p = head_->forward[0]; p; p = p->forward[0]) {
std::cout << p->key << " ";
}
std::cout << "\n";
}
private:
int random_level() {
int lv = 1;
while ((rand() % 100) < 50 && lv < MAX_LEVEL) lv++; // 50% 概率再升一层
return lv;
}
Node* head_;
int level_;
int size_;
};
int main() {
SkipList sl;
// 注意：key 用字符串比大小，"3" > "29"（字典序），所以补零对齐
std::string keys[10] = {"51", "03", "88", "12", "67", "29", "95", "40", "07", "73"};
for (int i = 0; i < 10; i++) {
sl.put(keys[i], "score_" + keys[i]);
}
std::cout << "乱序插入 10 个学号完成，共 " << sl.size() << " 条\n\n";
sl.print_all();
std::cout << "\n";
sl.print_levels();
std::cout << "\n";
sl.trace_get("88");
std::string v;
if (sl.get("07", v)) std::cout << "\nget(\"07\") = " << v << "\n";
if (!sl.get("66", v)) std::cout << "get(\"66\") 没找到（正确）\n";
return 0;
}这课的新语法是指针和结构体。struct Node 和 class 几乎一样，区别只是成员默认公开，我们把简单的"数据袋子"叫 struct。Node* 是指针——存的是另一个 Node 的门牌号（内存地址），不是 Node 本身。nullptr 是"空指针"，表示"这里没有下一站"。new Node(...) 在内存里造一个新结点，返回它的门牌号；delete p 把门牌号指向的内存还回去——new 和 delete 必须配对，只 new 不 delete 叫内存泄漏，程序越跑越肥。析构函数 ~SkipList() 在对象销毁时自动执行，我们在这里把所有结点逐个 delete。
跳表的原理概念篇讲过：底层是一条有序链表（第 0 层），上面按"抛硬币"盖快车道。forward 是个指针数组，forward[i] 就是"我在第 i 层的下一站"。查找从最高层出发，本层下一站还比目标小就跳过去，否则下到下一层——像坐高铁，先快车后慢车。put 里那个 update 数组是唯一难懂的地方：找插入位置时，每一层"跳到最后一个还没超线的结点"时顺手记下它在哪（update[lv] = cur），新结点造好后逐层插到 update[i] 和它下一站之间——先接新结点的线，再改旧结点的线，两行代码顺序不能反。
编译运行：
g++ -std=c++17 -O2 skiplist.cpp -o skiplist
./skiplist你应该看到（层数分配是随机的，但你这里因为固定了种子，会长成和我一模一样的样子）：
乱序插入 10 个学号完成，共 10 条
第 0 层从头到尾（已有序）: 03 07 12 29 40 51 67 73 88 95
第 3 层: 29
第 2 层: 29 73
第 1 层: 03 29 73 95
第 0 层: 03 07 12 29 40 51 67 73 88 95
查找 88 的路线:
第 3 层跳到 29
第 2 层跳到 73
第 0 层命中: 88 = score_88
get("07") = score_07
get("66") 没找到（正确）乱序进去，有序出来。查找 88：第 3 层从出头跳到 29，29 的下一站 73 还是比 88 小？不，73 比 88 小所以跳到 73——看输出确认路线；73 在第 2 层的下一站是空，降到第 1 层，再降第 0 层命中 88。10 个元素要走 3 跳，100 万个元素平均只要约 20 跳（层数是 log 级的）。
注意主函数里那行注释：字符串比大小是字典序，"29" < "3"（因为 '2' < '3'），所以数字 key 要补零对齐（"03"）。真实存储引擎对 key 的编码排序有很多讲究，这是入门第一课。
调试 忘写析构函数程序也能跑（操作系统会回收），但那是坏习惯；forward 忘了初始化长度就访问 forward[lv] 会段错误（Segmentation Fault，程序崩溃的名字，意思是"访问了不该访问的内存"）；new 出来的结点如果没人 delete，有些工具（如 valgrind）会给你列出每一处泄漏。
练习 给 SkipList 加一个 min_key() 函数，返回第 0 层第一个结点的 key（head_->forward[0]->key），空表返回空字符串。
M4第五课 M4：SSTable 落盘 + 稀疏索引
目标 MemTable 攒满 100 条自动冻结、按序写成一个 SSTable 文件，并在内存里给每个文件建"每 100 条一个路标"的稀疏索引。get 变成"内存没有就带着路标去磁盘，只读一小段"。对应简历词条：WAL、Mutable/Immutable MemTable、多级 SSTable。
新建 m4_sstable/sst_kv.cpp。这个文件是 M2 的加厚版，完整代码如下：
#include <iostream>
#include <fstream>
#include <map>
#include <string>
#include <vector>
const int MEMTABLE_LIMIT = 100; // MemTable 攒够 100 条就落盘（教学用小数字，真实引擎按 MB 算）
const int INDEX_STEP = 100; // 稀疏索引：每 100 条立一个路标
// MiniKV：WAL + MemTable + SSTable + 稀疏索引
class MiniKV {
public:
MiniKV() {
std::ifstream in("wal.log");
std::string k, v;
while (in >> k >> v) memtable_[k] = v;
wal_.open("wal.log", std::ios::app);
next_id_ = 1;
}
void put(const std::string& key, const std::string& value) {
wal_ << key << " " << value << "\n";
wal_.flush();
memtable_[key] = value;
if ((int)memtable_.size() >= MEMTABLE_LIMIT) flush();
}
bool get(const std::string& key, std::string& value) {
// 路线 1：先查内存（最新）
std::map<std::string, std::string>::iterator it = memtable_.find(key);
if (it != memtable_.end()) {
value = it->second;
last_path_ = "MemTable";
return true;
}
// 路线 2：从新到旧，一个一个查 SSTable
for (size_t i = 0; i < ssts_.size(); i++) {
if (get_from_sst(ssts_[i], key, value)) {
last_path_ = ssts_[i].filename;
return true;
}
}
return false;
}
std::string last_path() const { return last_path_; }
private:
// 一个 SSTable = 一个数据文件 + 一张稀疏索引
struct SSTable {
std::string filename;
std::map<std::string, long long> index; // 路标 key -> 文件偏移量
};
void flush() {
SSTable s;
s.filename = "sst_" + std::to_string(next_id_) + ".data";
next_id_++;
std::ofstream out(s.filename.c_str(), std::ios::binary);
long long offset = 0; // 当前写到了文件的第几个字节
int cnt = 0;
// map 本身有序，直接顺序写出去，文件就是有序的
for (std::map<std::string, std::string>::iterator it = memtable_.begin();
it != memtable_.end(); ++it) {
if (cnt % INDEX_STEP == 0) s.index[it->first] = offset; // 立路标
out << it->first << " " << it->second << "\n";
offset += (long long)(it->first.size() + 1 + it->second.size() + 1);
cnt++;
}
out.close(); // 写完立刻关闭，数据才算真正落盘
ssts_.insert(ssts_.begin(), s); // 最新的 SSTable 放最前面
memtable_.clear();
// MemTable 的数据已经安全落盘，WAL 可以清空重来了
wal_.close();
wal_.open("wal.log", std::ios::trunc);
std::cout << "[flush] MemTable 满 " << MEMTABLE_LIMIT
<< " 条，写成了 " << s.filename << "\n";
}
bool get_from_sst(const SSTable& s, const std::string& key, std::string& value) {
if (s.index.empty()) return false;
// 找到 <= key 的最后一个路标
std::map<std::string, long long>::const_iterator it = s.index.upper_bound(key);
if (it == s.index.begin()) return false; // 比最小路标还小，肯定不在
--it;
std::ifstream in(s.filename.c_str(), std::ios::binary);
in.seekg(it->second);
std::string k, v;
int scanned = 0;
while (in >> k >> v) {
scanned++;
if (k == key) { value = v; return true; }
if (k > key) break; // 文件有序，走过头了 = 没这个 key
if (scanned > INDEX_STEP + 2) break; // 保险丝
}
return false;
}
std::map<std::string, std::string> memtable_;
std::ofstream wal_;
std::vector<SSTable> ssts_;
int next_id_;
std::string last_path_;
};
int main() {
MiniKV db;
for (int i = 0; i < 250; i++) {
db.put("user:" + std::to_string(i), "likes_" + std::to_string(i % 50));
}
std::string v;
if (db.get("user:37", v))
std::cout << "user:37 -> 在 " << db.last_path() << " 找到, 值=" << v << "\n";
if (db.get("user:150", v))
std::cout << "user:150 -> 在 " << db.last_path() << " 找到, 值=" << v << "\n";
if (db.get("user:230", v))
std::cout << "user:230 -> 在 " << db.last_path() << " 找到, 值=" << v << "\n";
return 0;
}核心机制讲三处。第一，flush() 是 Immutable MemTable 的化身：MemTable 满 100 条就整体按序写进 sst_N.data，然后 memtable_.clear() 清空——概念篇说的"满了就冻结成不可变文件"就这么朴素。写文件时随手干两件事：每 100 条记一个路标（s.index[it->first] = offset，路标是 key 到文件偏移量的映射），顺便算好每条数据占几个字节（key 长度 + 1 个空格 + value 长度 + 1 个换行）。写完 out.close() ——这个 close 千万不能少，后面有我亲身调试的血泪故事。
第二，get_from_sst 是"先定位再小范围翻"：upper_bound(key) 在路标表里找到第一个比 key 大的路标，往回退一个就是"不大于 key 的最后一个路标"，它的 offset 就是我们该跳到的位置。in.seekg(offset) 把读文件的位置直接拨到那里——前面的字节一个都不用读。然后最多扫 100 来行，看到 key 相等就命中，看到比 key 大的就知道没戏了（文件有序）。对比 M1 的全文件扫描：M1 扫 19500 行，这里扫几十行。
第三，ssts_ 是个 vector，ssts_.insert(ssts_.begin(), s) 把新文件插到最前面——所以数组里越靠前越新。get 时从下标 0（最新）往旧里查，保证拿到的是最新版本。
编译运行（先删干净旧文件）：
g++ -std=c++17 -O2 sst_kv.cpp -o sst_kv
rm -f wal.log sst_*.data
./sst_kv你应该看到：
[flush] MemTable 满 100 条，写成了 sst_1.data
[flush] MemTable 满 100 条，写成了 sst_2.data
user:37 -> 在 sst_1.data 找到, 值=likes_37
user:150 -> 在 sst_2.data 找到, 值=likes_0
user:230 -> 在 MemTable 找到, 值=likes_30250 条数据被切成了三截：sst_1 装最老的 100 条，sst_2 装中间 100 条，MemTable 里还剩 50 条最新的。三个查询分别命中三种不同的存储位置——这就是 LSM 读路径的雏形。
调试现场一（我亲身踩的坑，你必须知道）：我第一版代码打开文件没加 std::ios::binary，在 Windows 上跑出来的结果忽对忽错，user:250 怎么都查不到。折腾半天的结论是：Windows 默认用文本模式打开文件，程序里写的 \n（换行，1 个字节）落盘时被偷偷换成了 \r\n（回车换行，2 个字节）。我手工计算的偏移量全按 1 字节算，文件实际却多出了一堆隐藏字节，seekg 全部跳歪。教训：凡是涉及"自己算字节偏移量"的文件读写，一律用二进制模式打开（std::ios::binary），真实存储引擎无一例外。
调试 忘了 out.close() 就去读同一个文件，会读到半截（下一课细讲）；seekg 之前忘了判断 index.empty()，空文件会出错；路标查找方向搞反（用了 lower_bound），会定位到错误的段落。
练习 把 MEMTABLE_LIMIT 改成 30、INDEX_STEP 改成 10，重跑看 flush 次数和路标数量怎么变。再想想：INDEX_STEP 设成 1 会怎样？（路标遍地都是，定位极快但索引本身巨大——这就是空间换时间。）
M5第六课 M5：布隆过滤器，不存在的 key 一秒判死
目标 给每个 SSTable 挂一个手写布隆过滤器。查 100 个不存在的 key，布隆把绝大多数磁盘白跑直接拦下，程序自己把拦截次数报给你看。对应简历词条：Bloom Filter。
新建 m5_bloom/bloom_kv.cpp。类结构和 M4 相同，多了一个 BloomFilter 类和统计：
#include <iostream>
#include <fstream>
#include <map>
#include <string>
#include <vector>
#include <cstdint>
const int MEMTABLE_LIMIT = 100;
const int INDEX_STEP = 100;
// ===================== 布隆过滤器 =====================
// 底层是一个 64000 位的位数组（存在 1000 个 uint64 里）
class BloomFilter {
public:
static constexpr size_t NUM_BITS = 64000;
BloomFilter() : bits_(NUM_BITS / 64, 0) {}
void add(const std::string& key) {
uint64_t h1 = hash(key, 1469598103934665603ULL);
uint64_t h2 = hash(key, 14695981039346656037ULL);
set_bit(h1 % NUM_BITS);
set_bit(h2 % NUM_BITS);
}
bool may_contain(const std::string& key) const {
uint64_t h1 = hash(key, 1469598103934665603ULL);
uint64_t h2 = hash(key, 14695981039346656037ULL);
return get_bit(h1 % NUM_BITS) && get_bit(h2 % NUM_BITS);
}
private:
// FNV-1a 哈希：basis 不同就是两个不同的哈希函数
static uint64_t hash(const std::string& s, uint64_t basis) {
uint64_t h = basis;
for (size_t i = 0; i < s.size(); i++) {
h ^= (unsigned char)s[i];
h *= 1099511628211ULL;
}
return h;
}
void set_bit(size_t pos) { bits_[pos / 64] |= (1ULL << (pos % 64)); }
bool get_bit(size_t pos) const { return (bits_[pos / 64] >> (pos % 64)) & 1ULL; }
std::vector<uint64_t> bits_;
};
// ===================== KV 引擎（带布隆） =====================
class MiniKV {
public:
MiniKV() {
std::ifstream in("wal.log");
std::string k, v;
while (in >> k >> v) memtable_[k] = v;
wal_.open("wal.log", std::ios::app);
next_id_ = 1;
}
void put(const std::string& key, const std::string& value) {
wal_ << key << " " << value << "\n";
wal_.flush();
memtable_[key] = value;
if ((int)memtable_.size() >= MEMTABLE_LIMIT) flush();
}
bool get(const std::string& key, std::string& value) {
std::map<std::string, std::string>::iterator it = memtable_.find(key);
if (it != memtable_.end()) {
value = it->second;
last_path_ = "MemTable";
return true;
}
for (size_t i = 0; i < ssts_.size(); i++) {
if (ssts_[i].bloom.may_contain(key)) {
// 布隆说"可能有"，才值得打开文件
if (get_from_sst(ssts_[i], key, value)) {
last_path_ = ssts_[i].filename + " (布隆放行)";
return true;
}
} else {
bloom_blocked_++; // 布隆说"绝对没有"，文件一次都不用打开
}
}
return false;
}
std::string last_path() const { return last_path_; }
long long bloom_blocked() const { return bloom_blocked_; }
private:
struct SSTable {
std::string filename;
std::map<std::string, long long> index;
BloomFilter bloom; // 每个 SSTable 自带一个布隆过滤器
};
void flush() {
SSTable s;
s.filename = "sst_" + std::to_string(next_id_) + ".data";
next_id_++;
std::ofstream out(s.filename.c_str(), std::ios::binary);
long long offset = 0;
int cnt = 0;
for (std::map<std::string, std::string>::iterator it = memtable_.begin();
it != memtable_.end(); ++it) {
if (cnt % INDEX_STEP == 0) s.index[it->first] = offset;
s.bloom.add(it->first); // 顺手把 key 塞进布隆过滤器
out << it->first << " " << it->second << "\n";
offset += (long long)(it->first.size() + 1 + it->second.size() + 1);
cnt++;
}
out.close();
ssts_.insert(ssts_.begin(), s);
memtable_.clear();
wal_.close();
wal_.open("wal.log", std::ios::trunc);
std::cout << "[flush] 写成 " << s.filename
<< "（索引 " << s.index.size() << " 个路标，布隆已建好）\n";
}
bool get_from_sst(const SSTable& s, const std::string& key, std::string& value) {
if (s.index.empty()) return false;
std::map<std::string, long long>::const_iterator it = s.index.upper_bound(key);
if (it == s.index.begin()) return false;
--it;
std::ifstream in(s.filename.c_str(), std::ios::binary);
in.seekg(it->second);
std::string k, v;
int scanned = 0;
while (in >> k >> v) {
scanned++;
if (k == key) { value = v; return true; }
if (k > key) break;
if (scanned > INDEX_STEP + 2) break;
}
return false;
}
std::map<std::string, std::string> memtable_;
std::ofstream wal_;
std::vector<SSTable> ssts_;
int next_id_;
std::string last_path_;
long long bloom_blocked_ = 0;
};
int main() {
MiniKV db;
for (int i = 0; i < 250; i++) {
db.put("user:" + std::to_string(i), "likes_" + std::to_string(i % 50));
}
std::string v;
// 100 个不存在的 key：布隆过滤器应该把绝大部分直接拦下
for (int q = 0; q < 100; q++) {
db.get("ghost:" + std::to_string(q), v);
}
std::cout << "查 100 个不存在的 key，布隆过滤器拦下了 "
<< db.bloom_blocked() << " 次磁盘白跑\n";
if (db.get("user:37", v))
std::cout << "user:37 -> 在 " << db.last_path() << " 找到\n";
return 0;
}新语法先过一遍。#include <cstdint> 引入 uint64_t（无符号 64 位整数）这些定宽类型。static constexpr size_t NUM_BITS = 64000 是"类级别的常量"，constexpr 表示"编译期就算好"。std::vector<uint64_t> bits_(NUM_BITS / 64, 0) 造了 1000 个初值为 0 的 64 位整数，拼成一个 64000 位的位数组——我们用一个整数的第 k 位存 0 或 1，64000 个开关只花 8KB 内存。位操作三件套：set_bit(pos) 把第 pos 位设成 1，pos / 64 算出是第几个整数，1ULL << (pos % 64) 造出一个"只有这一位是 1"的掩码，用 |=（按位或）盖上去；get_bit(pos) 用 >> 把目标位挪到最低位再 & 1ULL 看是 0 是 1。h ^= byte 是按位异或，FNV-1a 就是"异或一个字节、乘一个素数"反复滚动，两个不同的初始 basis 造出两个互不相干的哈希函数。
逻辑规则一句话：插入时把 key 的两个哈希位都拨成 1；查询时看这两位是否都还是 1。只要有一位是 0，说明从来没有任何 key 碰过这两位——这个 key 绝对不存在，磁盘都不用开。两位都是 1 才"可能存在"（也可能是别的 key 碰巧占的），这时才值得开文件。所以布隆过滤器只会漏报"存在"（假阳性），绝不会错杀"不存在"——错杀是灾难（数据丢了），漏报只是多开一次文件，无伤大雅。
编译运行：
g++ -std=c++17 -O2 bloom_kv.cpp -o bloom_kv
rm -f wal.log sst_*.data
./bloom_kv你应该看到：
[flush] 写成 sst_1.data（索引 1 个路标，布隆已建好）
[flush] 写成 sst_2.data（索引 1 个路标，布隆已建好）
查 100 个不存在的 key，布隆过滤器拦下了 200 次磁盘白跑
user:37 -> 在 sst_1.data (布隆放行) 找到100 个幽灵 key，每个都要问两个 SSTable，共 200 次提问，200 次全被拦下——没有一次打开文件。user:37 真实存在，布隆放行，顺利在 sst_1 找到。没有布隆的 M4 里，这 200 次提问就是 200 次开文件加扫描。
调试 1 << (pos % 64) 忘了写 1ULL 会出大事——1 默认是 32 位 int，左移超过 31 位就溢出出错，这是位运算新手第二坑（第一坑是忘了位编号从 0 开始）。哈希函数只用一个的话，碰撞率会高到假阳性频发，所以要用两个。
练习 把 NUM_BITS 从 64000 减到 1000，重跑看拦截次数还剩多少——位数太少碰撞就多，假阳性就高。这让你亲手体会"布隆过滤器的大小决定误判率"。
M6第七课 M6：墓碑与 Compaction，删除的艺术
目标 删除不再是真的抹掉，而是写一个墓碑标记；SSTable 超过 3 个自动合并成一个，去重、扔墓碑，旧文件删掉。对应简历词条：Compaction 策略、空间回收。
先想清楚一个反直觉的事实：LSM 里删除一个 key，不可能冲到磁盘上把那行抹掉——数据散落在不可变的 SSTable 里，找到它可能要翻好几个文件，抹掉还会留下一个洞。所以做法是：往 MemTable 里写一个特殊值 __TOMBSTONE__（墓碑），表示"这个 key 死了"。以后 get 查到墓碑，就当作不存在。墓碑什么时候真正消失？Compaction 合并重写文件的时候——新文件里不写它，它就永久消失了。
新建 m6_compaction/compaction_kv.cpp，类结构和 M5 相同，变化四处：多一个 TOMBSTONE 常量、多一个 del 函数、get 遇到墓碑返回不存在、多一个 compact 函数：
#include <iostream>
#include <fstream>
#include <cstdio>
#include <map>
#include <string>
#include <vector>
#include <cstdint>
const int MEMTABLE_LIMIT = 100;
const int INDEX_STEP = 100;
const std::string TOMBSTONE = "__TOMBSTONE__"; // 墓碑：删除标记
class BloomFilter {
public:
static constexpr size_t NUM_BITS = 64000;
BloomFilter() : bits_(NUM_BITS / 64, 0) {}
void add(const std::string& key) {
uint64_t h1 = hash(key, 1469598103934665603ULL);
uint64_t h2 = hash(key, 14695981039346656037ULL);
set_bit(h1 % NUM_BITS);
set_bit(h2 % NUM_BITS);
}
bool may_contain(const std::string& key) const {
uint64_t h1 = hash(key, 1469598103934665603ULL);
uint64_t h2 = hash(key, 14695981039346656037ULL);
return get_bit(h1 % NUM_BITS) && get_bit(h2 % NUM_BITS);
}
private:
static uint64_t hash(const std::string& s, uint64_t basis) {
uint64_t h = basis;
for (size_t i = 0; i < s.size(); i++) {
h ^= (unsigned char)s[i];
h *= 1099511628211ULL;
}
return h;
}
void set_bit(size_t pos) { bits_[pos / 64] |= (1ULL << (pos % 64)); }
bool get_bit(size_t pos) const { return (bits_[pos / 64] >> (pos % 64)) & 1ULL; }
std::vector<uint64_t> bits_;
};
class MiniKV {
public:
MiniKV() {
std::ifstream in("wal.log");
std::string k, v;
while (in >> k >> v) memtable_[k] = v;
wal_.open("wal.log", std::ios::app);
next_id_ = 1;
}
void put(const std::string& key, const std::string& value) {
wal_ << key << " " << value << "\n";
wal_.flush();
memtable_[key] = value;
if ((int)memtable_.size() >= MEMTABLE_LIMIT) flush();
}
// 删除 = 写一个墓碑。数据并没有真的消失！
void del(const std::string& key) {
put(key, TOMBSTONE);
}
bool get(const std::string& key, std::string& value) {
std::map<std::string, std::string>::iterator it = memtable_.find(key);
if (it != memtable_.end()) {
if (it->second == TOMBSTONE) { last_path_ = "MemTable(墓碑)"; return false; }
value = it->second;
last_path_ = "MemTable";
return true;
}
for (size_t i = 0; i < ssts_.size(); i++) {
if (ssts_[i].bloom.may_contain(key)) {
if (get_from_sst(ssts_[i], key, value)) {
if (value == TOMBSTONE) { last_path_ = ssts_[i].filename + "(墓碑)"; return false; }
last_path_ = ssts_[i].filename;
return true;
}
}
}
return false;
}
private:
struct SSTable {
std::string filename;
std::map<std::string, long long> index;
BloomFilter bloom;
};
void flush() {
SSTable s;
s.filename = "sst_" + std::to_string(next_id_) + ".data";
next_id_++;
std::ofstream out(s.filename.c_str(), std::ios::binary);
long long offset = 0;
int cnt = 0;
for (std::map<std::string, std::string>::iterator it = memtable_.begin();
it != memtable_.end(); ++it) {
if (cnt % INDEX_STEP == 0) s.index[it->first] = offset;
s.bloom.add(it->first);
out << it->first << " " << it->second << "\n";
offset += (long long)(it->first.size() + 1 + it->second.size() + 1);
cnt++;
}
out.close(); // 关键！不 close 的话数据还躺在缓冲区里，compact 读到的是空文件
ssts_.insert(ssts_.begin(), s);
memtable_.clear();
wal_.close();
wal_.open("wal.log", std::ios::trunc);
std::cout << "[flush] 写成 " << s.filename << "\n";
if (ssts_.size() > 3) compact(); // SSTable 超过 3 个就合并
}
// Compaction：从新到旧读所有 SSTable，新的 key 说了算，墓碑直接扔掉
void compact() {
std::cout << "[compact] 有 " << ssts_.size() << " 个 SSTable，开始合并\n";
std::map<std::string, std::string> merged;
std::map<std::string, bool> deleted; // 记住"哪些 key 已删除"，防止老文件的旧值复活
for (size_t i = 0; i < ssts_.size(); i++) { // 越靠前越新
std::ifstream in(ssts_[i].filename.c_str(), std::ios::binary);
std::string k, v;
while (in >> k >> v) {
if (merged.count(k) || deleted.count(k)) continue; // 已有更新版本，跳过
if (v == TOMBSTONE) {
deleted[k] = true; // 标记删除，老文件的旧值不许再进来
continue;
}
merged[k] = v;
}
}
SSTable s;
s.filename = "sst_" + std::to_string(next_id_) + ".data";
next_id_++;
std::ofstream out(s.filename.c_str(), std::ios::binary);
long long offset = 0;
int cnt = 0;
for (std::map<std::string, std::string>::iterator it = merged.begin();
it != merged.end(); ++it) {
if (cnt % INDEX_STEP == 0) s.index[it->first] = offset;
s.bloom.add(it->first);
out << it->first << " " << it->second << "\n";
offset += (long long)(it->first.size() + 1 + it->second.size() + 1);
cnt++;
}
out.close(); // 写完立刻刷盘
std::cout << "[compact] 合并完成: " << merged.size()
<< " 条有效数据 -> " << s.filename << "\n";
// 旧文件已经没用了，从磁盘上删掉（不删的话磁盘越攒越多）
for (size_t i = 0; i < ssts_.size(); i++) {
std::remove(ssts_[i].filename.c_str());
}
ssts_.clear();
ssts_.push_back(s);
}
bool get_from_sst(const SSTable& s, const std::string& key, std::string& value) {
if (s.index.empty()) return false;
std::map<std::string, long long>::const_iterator it = s.index.upper_bound(key);
if (it == s.index.begin()) return false;
--it;
std::ifstream in(s.filename.c_str(), std::ios::binary);
in.seekg(it->second);
std::string k, v;
int scanned = 0;
while (in >> k >> v) {
scanned++;
if (k == key) { value = v; return true; }
if (k > key) break;
if (scanned > INDEX_STEP + 2) break;
}
return false;
}
std::map<std::string, std::string> memtable_;
std::ofstream wal_;
std::vector<SSTable> ssts_;
int next_id_;
std::string last_path_;
};
int main() {
MiniKV db;
for (int i = 0; i < 350; i++) {
db.put("user:" + std::to_string(i), "likes_" + std::to_string(i % 50));
}
for (int i = 0; i < 50; i++) {
db.del("user:" + std::to_string(i)); // 删掉 50 个
}
std::string v;
std::cout << "\n";
if (!db.get("user:10", v)) std::cout << "user:10 已删除（get 返回 false）\n";
if (!db.get("ghost:1", v)) std::cout << "ghost:1 不存在（get 返回 false）\n";
if (db.get("user:250", v)) std::cout << "user:250 还在, 值=" << v << "\n";
return 0;
}compact 的算法值得逐句抠。合并从最新的 SSTable 读到最旧的（for 循环 i 从 0 开始，ssts_ 越靠前越新），读到 key 先问两个问题：merged 里已经有它了吗？deleted 里有它吗？有任何一个，跳过——因为更年轻的版本已经发过话了。遇到墓碑，记进 deleted 后跳过。等全部读完，merged 里就是"每个 key 还活着的最新版本"，墓碑的信息转化成了 deleted 里的一个名字，新旧文件里它都没了。最后写新文件、std::remove 删旧文件（需要 #include <cstdio>）。
调试现场二（这条是全篇最有价值的 bug，认真看）：我第一版 compact 写的是"遇到墓碑就 continue，不进 merged"，结果测试时 user:10 删掉后居然还能查到！推理一遍就明白了：sst_4 里有墓碑 user:10，我跳过它；继续读更老的 sst_1，里面有 user:10 的旧值 likes_10，merged 里没有它、也没有人拦着它，于是旧值复活了。修复就是代码里那张 deleted 表：墓碑不光自己不进新文件，还要封杀所有更老的同名 key。第二个 bug 更阴险：合并函数读 sst_4 时读到了空文件。原因是我把"写完文件"和"合并"写在了同一个函数调用栈里，写文件的 ofstream 还没析构，数据全在 4KB 的缓冲区里没落盘，磁盘上的 sst_4 是个空壳。修复是写完立刻 out.close()。同一批数据"写的人还没写完，读的人就开始读"，这是并发世界的经典事故，存储引擎里管这叫可见性问题。
编译运行：
g++ -std=c++17 -O2 compaction_kv.cpp -o compaction_kv
rm -f wal.log sst_*.data
./compaction_kv你应该看到：
[flush] 写成 sst_1.data
[flush] 写成 sst_2.data
[flush] 写成 sst_3.data
[flush] 写成 sst_4.data
[compact] 有 4 个 SSTable，开始合并
[compact] 合并完成: 300 条有效数据 -> sst_5.data
user:10 已删除（get 返回 false）
ghost:1 不存在（get 返回 false）
user:250 还在, 值=likes_0350 条 put + 50 条 del，四个文件合并后只剩 300 条有效数据，旧文件全部删除，目录里只有 sst_5.data。被删的 user:10 查不到了，没删的 user:250 活得好好的。
调试 deleted 表忘了做，删除的 key 会"复活"；std::remove 需要 cstdio 头文件；合并时忘了从新到旧的顺序（从旧往新读），所有 key 都会变成最老的版本。
练习 把触发阈值从 3 改成 2，重跑看 compact 提前发生。再想一个问题：如果删除之后永远不触发 compaction，磁盘上会发生什么？（墓碑和旧数据永远堆着——所以真实引擎有"强制全量合并"的保底手段。）
M7第八课 M7：毕业整合——跳表 MemTable 的完整 LSM 引擎
目标 把前六课全部组装进一个类：跳表 MemTable + WAL + SSTable 稀疏索引 + 布隆过滤器 + 墓碑 + Compaction，然后灌 2000 条抖音点赞记录、删 200 条、查 500 次，打一张"毕业成绩单"。这份代码就是你能写进简历的东西。
下面把 430 行拆成 4 段讲，先看讲解，再看对应代码片段。文件还是 m7_final_engine/mini_lsm.cpp，敲的时候分三次：先 SkipList、再 BloomFilter、最后 MiniLSM。
先画一张「总装图」：六件套怎么焊在一起
六件套不是散装零件，而是按一条读写流水线焊死的。先把这张图（文字版）刻进脑子：
写入请求 put(k, v)
│
▼
① WAL 写前日志 ── 先记一笔到磁盘，断电也不丢
│
▼
② 跳表 MemTable（内存，有序）── 最新数据都在这
│ MemTable 满了（≥ MEMTABLE_LIMIT）
▼
③ flush 落盘：跳表 collect 成有序序列
├─ 写 SSTable 文件（磁盘上的有序表）
├─ ④ 建稀疏索引（每 INDEX_STEP 条一个路标）
└─ ⑤ 建布隆过滤器（每个 key 占两个 bit）
│ SSTable 超过 3 个
▼
⑥ Compaction：多路归并，去重 + 丢墓碑 + 压成 1 个
读取请求 get(k)
│
▼
① 先问 MemTable（最快，最新）
│ 没有
▼
② 从新到旧遍历每个 SSTable
├─ ③ 先问布隆：说"绝对没有"就跳过磁盘
├─ ④ 稀疏索引定位到大致位置
├─ ⑤ 小范围扫描几行
└─ ⑥ 遇到墓碑(__TOMBSTONE__)返回"不存在"一句话总结：写时层层加码防丢，读时层层短路防慢。下面四段代码，正好对应这张图的左右两半——别再一口气吞 430 行，一段一段吃。
#### 第 1 段：头文件与常量
开头是头文件和三个常量。MEMTABLE_LIMIT = 500 是 MemTable 的容量上限——写满就触发 flush 落盘（真实引擎按 MB 算，我们故意用小数字好演示节奏）。INDEX_STEP = 100 是稀疏索引的步长，每写 100 条立一个"路标"，查的时候先跳到最近路标再小范围扫。TOMBSTONE = "__TOMBSTONE__" 是墓碑标记——删除时不真删，而是写进这个特殊值，等 compaction 再物理清除（这就是 LSM"删除快、但后台要还债"的本质）。
#include <iostream>
#include <fstream>
#include <cstdio>
#include <map>
#include <string>
#include <vector>
#include <cstdint>
#include <utility>
#include <chrono>
#include <cstdlib>
const int MEMTABLE_LIMIT = 500; // MemTable 容量（教学用小数字，真实引擎按 MB 算）
const int INDEX_STEP = 100; // 稀疏索引：每 100 条立一个路标
const std::string TOMBSTONE = "__TOMBSTONE__";
#### 第 2 段：跳表 SkipList（M3 原班人马，+collect +clear +禁拷贝）
跳表是 M3 的原班人马，这里只多了三样东西，也是毕业整合的关键：
collect()：把跳表里所有 (key, value) 沿第 0 层从头走到尾收集成一个有序vector——跳表本来就有序，这只是一路走到头，落盘时直接顺序写出即可。clear()：落盘后把跳表清空换成空表。注意不能直接memtable_ = SkipList()——跳表里是裸指针，C++ 默认拷贝只复制指针不复制内存，等号右边的临时对象析构时会把那块内存delete掉，左边对象手里的指针立刻变成野指针，一碰就段错误。这就是 C++ 的三法则：自己管了new出来的资源，就必须自己写析构 / 拷贝构造 / 拷贝赋值，或者像我们这样直接= delete禁掉拷贝，改用clear()。这个坑你写到这里大概率会踩，踩了正好按下面"调试现场"修。- 析构函数
~SkipList()：顺着第 0 层把每个结点delete掉，防止内存泄漏。
// 第一件套：跳表 SkipList（M3 的原班人马，加了一个 collect）
const int MAX_LEVEL = 8;
struct Node {
std::string key;
std::string value;
std::vector<Node*> forward;
Node(const std::string& k, const std::string& v, int level)
: key(k), value(v), forward(level, nullptr) {}
};
class SkipList {
public:
SkipList() : head_(new Node("", "", MAX_LEVEL)), level_(1), size_(0) {
srand(42);
}
~SkipList() {
Node* p = head_->forward[0];
while (p) {
Node* nxt = p->forward[0];
delete p;
p = nxt;
}
delete head_;
}
int size() const { return size_; }
// 清空所有结点，回到空表状态（MemTable 落盘后用）
void clear() {
Node* p = head_->forward[0];
while (p) {
Node* nxt = p->forward[0];
delete p;
p = nxt;
}
for (int i = 0; i < MAX_LEVEL; i++) head_->forward[i] = nullptr;
level_ = 1;
size_ = 0;
}
// 禁止拷贝：内部是裸指针，默认拷贝会把同一块内存删两次（三法则！）
SkipList(const SkipList&) = delete;
SkipList& operator=(const SkipList&) = delete;
bool put(const std::string& k, const std::string& v) {
std::vector<Node*> update(MAX_LEVEL, head_);
Node* cur = head_;
for (int lv = level_ - 1; lv >= 0; lv--) {
while (cur->forward[lv] && cur->forward[lv]->key < k) {
cur = cur->forward[lv];
}
update[lv] = cur;
}
Node* next = cur->forward[0];
if (next && next->key == k) {
next->value = v;
return false;
}
int lv = random_level();
if (lv > level_) {
for (int i = level_; i < lv; i++) update[i] = head_;
level_ = lv;
}
Node* node = new Node(k, v, lv);
for (int i = 0; i < lv; i++) {
node->forward[i] = update[i]->forward[i];
update[i]->forward[i] = node;
}
size_++;
return true;
}
bool get(const std::string& k, std::string& v) {
Node* cur = head_;
for (int lv = level_ - 1; lv >= 0; lv--) {
while (cur->forward[lv] && cur->forward[lv]->key < k) {
cur = cur->forward[lv];
}
}
Node* next = cur->forward[0];
if (next && next->key == k) {
v = next->value;
return true;
}
return false;
}
// 新增：把所有 (key, value) 按有序顺序收集出来，落盘时用
void collect(std::vector<std::pair<std::string, std::string> >& out) const {
for (Node* p = head_->forward[0]; p; p = p->forward[0]) {
out.push_back(std::make_pair(p->key, p->value));
}
}
private:
int random_level() {
int lv = 1;
while ((rand() % 100) < 50 && lv < MAX_LEVEL) lv++;
return lv;
}
Node* head_;
int level_;
int size_;
};
#### 第 3 段：布隆过滤器（M5 原班人马）
布隆过滤器是 M5 的原班人马，原样搬来：两个哈希函数各算一个 bit 位，may_contain 返回 true 表示"可能在"，false 表示"绝对不在"——没有假阴性，只有假阳性。读路径上全靠它帮我们跳过根本不存在的 key，省掉无谓的磁盘读取。
// 第二件套：布隆过滤器（M5 的原班人马）
class BloomFilter {
public:
static constexpr size_t NUM_BITS = 64000;
BloomFilter() : bits_(NUM_BITS / 64, 0) {}
void add(const std::string& key) {
uint64_t h1 = hash(key, 1469598103934665603ULL);
uint64_t h2 = hash(key, 14695981039346656037ULL);
set_bit(h1 % NUM_BITS);
set_bit(h2 % NUM_BITS);
}
bool may_contain(const std::string& key) const {
uint64_t h1 = hash(key, 1469598103934665603ULL);
uint64_t h2 = hash(key, 14695981039346656037ULL);
return get_bit(h1 % NUM_BITS) && get_bit(h2 % NUM_BITS);
}
private:
static uint64_t hash(const std::string& s, uint64_t basis) {
uint64_t h = basis;
for (size_t i = 0; i < s.size(); i++) {
h ^= (unsigned char)s[i];
h *= 1099511628211ULL;
}
return h;
}
void set_bit(size_t pos) { bits_[pos / 64] |= (1ULL << (pos % 64)); }
bool get_bit(size_t pos) const { return (bits_[pos / 64] >> (pos % 64)) & 1ULL; }
std::vector<uint64_t> bits_;
};
#### 第 4 段：MiniLSM 引擎本体（核心，上面走一遍读写路径）
MiniLSM 是引擎本体，对外只暴露 put / del / get。构造时先重放 WAL 做崩溃恢复（把 wal.log 每行读回 MemTable），再以追加模式打开 WAL——这就是为什么断电重启数据还在。
写路径（put）：① 把 k v 追加进 WAL 并 flush()，确保先落盘；② 写进跳表 MemTable；③ 若 MemTable 达到 MEMTABLE_LIMIT 就调 flush()。
flush()：跳表 collect 出有序序列 → 写 SSTable 文件，同时每 INDEX_STEP 条记一个稀疏索引偏移、每个 key 加进布隆 → 把新 SSTable 插到 ssts_ 最前面（越靠前越新）→ clear() 清空 MemTable、truncate 截断 WAL（旧数据已落盘，WAL 可重来）→ 若 SSTable 超过 3 个就 compact()。
读路径（get）：① 先问 MemTable，命中直接返回（删过的 key 在 MemTable 里是墓碑，返回不存在）；② 从新到旧遍历每个 SSTable：先问布隆 may_contain，说"绝对没有"就 continue 跳过整个文件（连磁盘都不碰），否则 get_from_sst 用稀疏索引 upper_bound 定位到最近路标、seekg 跳过去再小范围扫描几行；同样遇墓碑返回不存在。
compact()：把所有 SSTable 多路归并（用 std::map 自动按 key 排序去重，已合并过的 key 跳过，墓碑 key 记进 deleted 直接丢弃）→ 写出一个新 SSTable → 删掉旧的、只留新的。这就是"写放大"的来源，也是 LSM 用空间换写入速度的本质。
最后 main() 灌 2000 条、删 200 条、查 500 次，打印"毕业成绩单"。注意 get 里的 mem_hits_ / sst_hits_ / bloom_blocked_ 三个计数器，成绩单全靠它们说话。
// 第三件套：MiniLSM 引擎本体
class MiniLSM {
public:
MiniLSM() : next_id_(1), bloom_blocked_(0), sst_hits_(0), mem_hits_(0) {
// 崩溃恢复：重放 WAL
std::ifstream in("wal.log");
std::string k, v;
while (in >> k >> v) memtable_.put(k, v);
wal_.open("wal.log", std::ios::app);
}
void put(const std::string& key, const std::string& value) {
wal_ << key << " " << value << "\n";
wal_.flush();
memtable_.put(key, value);
if (memtable_.size() >= MEMTABLE_LIMIT) flush();
}
void del(const std::string& key) {
put(key, TOMBSTONE);
}
bool get(const std::string& key, std::string& value) {
// 路线 1：跳表 MemTable（最新数据在这）
std::string v;
if (memtable_.get(key, v)) {
mem_hits_++;
if (v == TOMBSTONE) { last_path_ = "MemTable(墓碑)"; return false; }
value = v;
last_path_ = "MemTable";
return true;
}
// 路线 2：从新到旧查 SSTable，先问布隆过滤器
for (size_t i = 0; i < ssts_.size(); i++) {
if (!ssts_[i].bloom.may_contain(key)) {
bloom_blocked_++; // 布隆说绝对没有，磁盘一次都不碰
continue;
}
if (get_from_sst(ssts_[i], key, v)) {
sst_hits_++;
if (v == TOMBSTONE) { last_path_ = ssts_[i].filename + "(墓碑)"; return false; }
value = v;
last_path_ = ssts_[i].filename;
return true;
}
}
return false;
}
std::string last_path() const { return last_path_; }
long long bloom_blocked() const { return bloom_blocked_; }
long long sst_hits() const { return sst_hits_; }
long long mem_hits() const { return mem_hits_; }
private:
struct SSTable {
std::string filename;
std::map<std::string, long long> index;
BloomFilter bloom;
};
void flush() {
SSTable s;
s.filename = "sst_" + std::to_string(next_id_) + ".data";
next_id_++;
std::vector<std::pair<std::string, std::string> > items;
memtable_.collect(items); // 跳表天然有序，直接顺序写出
std::ofstream out(s.filename.c_str(), std::ios::binary);
long long offset = 0;
for (size_t i = 0; i < items.size(); i++) {
if ((int)i % INDEX_STEP == 0) s.index[items[i].first] = offset;
s.bloom.add(items[i].first);
out << items[i].first << " " << items[i].second << "\n";
offset += (long long)(items[i].first.size() + 1 + items[i].second.size() + 1);
}
out.close(); // 必须先关文件再继续，否则缓冲区里的数据还没落盘
ssts_.insert(ssts_.begin(), s);
memtable_.clear(); // 换一个空跳表（不能直接赋值新 SkipList，见三法则）
wal_.close();
wal_.open("wal.log", std::ios::trunc);
std::cout << "[flush] 写成 " << s.filename << "\n";
if (ssts_.size() > 3) compact();
}
void compact() {
std::cout << "[compact] 有 " << ssts_.size() << " 个 SSTable，开始合并\n";
std::map<std::string, std::string> merged;
std::map<std::string, bool> deleted;
for (size_t i = 0; i < ssts_.size(); i++) { // 越靠前越新
std::ifstream in(ssts_[i].filename.c_str(), std::ios::binary);
std::string k, v;
while (in >> k >> v) {
if (merged.count(k) || deleted.count(k)) continue;
if (v == TOMBSTONE) { deleted[k] = true; continue; }
merged[k] = v;
}
}
SSTable s;
s.filename = "sst_" + std::to_string(next_id_) + ".data";
next_id_++;
std::ofstream out(s.filename.c_str(), std::ios::binary);
long long offset = 0;
int cnt = 0;
for (std::map<std::string, std::string>::iterator it = merged.begin();
it != merged.end(); ++it) {
if (cnt % INDEX_STEP == 0) s.index[it->first] = offset;
s.bloom.add(it->first);
out << it->first << " " << it->second << "\n";
offset += (long long)(it->first.size() + 1 + it->second.size() + 1);
cnt++;
}
out.close();
std::cout << "[compact] 合并完成: " << merged.size()
<< " 条有效数据 -> " << s.filename << "\n";
for (size_t i = 0; i < ssts_.size(); i++) {
std::remove(ssts_[i].filename.c_str());
}
ssts_.clear();
ssts_.push_back(s);
}
bool get_from_sst(const SSTable& s, const std::string& key, std::string& value) {
if (s.index.empty()) return false;
std::map<std::string, long long>::const_iterator it = s.index.upper_bound(key);
if (it == s.index.begin()) return false;
--it;
std::ifstream in(s.filename.c_str(), std::ios::binary);
in.seekg(it->second);
std::string k, v;
int scanned = 0;
while (in >> k >> v) {
scanned++;
if (k == key) { value = v; return true; }
if (k > key) break;
if (scanned > INDEX_STEP + 2) break;
}
return false;
}
SkipList memtable_;
std::ofstream wal_;
std::vector<SSTable> ssts_;
int next_id_;
std::string last_path_;
long long bloom_blocked_;
long long sst_hits_;
long long mem_hits_;
};
// 毕业演示：灌 2000 条点赞记录，删 200 条，查 500 次
int main() {
MiniLSM db;
auto t0 = std::chrono::steady_clock::now();
for (int i = 0; i < 2000; i++) {
db.put("user:" + std::to_string(i), "video:" + std::to_string(i % 50));
}
for (int i = 0; i < 200; i++) {
db.del("user:" + std::to_string(i));
}
auto t1 = std::chrono::steady_clock::now();
int found = 0, deleted = 0, missing = 0;
for (int q = 0; q < 300; q++) { // 300 个还活着的 key
std::string v;
if (db.get("user:" + std::to_string(200 + q), v)) found++;
}
for (int q = 0; q < 100; q++) { // 100 个删掉的 key
std::string v;
if (!db.get("user:" + std::to_string(q), v)) deleted++;
}
for (int q = 0; q < 100; q++) { // 100 个从来没存在的 key
std::string v;
if (!db.get("ghost:" + std::to_string(q), v)) missing++;
}
auto t2 = std::chrono::steady_clock::now();
long long write_ms = std::chrono::duration_cast<std::chrono::milliseconds>(t1 - t0).count();
long long read_ms = std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1).count();
std::cout << "\n========== 毕业成绩单 ==========\n";
std::cout << "写入 2000 条 + 删除 200 条 耗时: " << write_ms << " 毫秒\n";
std::cout << "查询 500 次耗时: " << read_ms << " 毫秒\n";
std::cout << "活着的 key 找到: " << found << "/300\n";
std::cout << "删掉的 key 正确返回不存在: " << deleted << "/100\n";
std::cout << "没存在的 key 正确返回不存在: " << missing << "/100\n";
std::cout << "路径统计: MemTable 命中 " << db.mem_hits()
<< " 次, SSTable 命中 " << db.sst_hits() << " 次\n";
std::cout << "布隆过滤器拦下磁盘白跑: " << db.bloom_blocked() << " 次\n";
return 0;
}三处新增讲清楚。第一，SkipList 加了 collect：落盘时要把跳表里所有键值对按序拿出来，沿第 0 层从头走到尾就是有序序列，装进一个 std::vector<std::pair<std::string,std::string>>（pair 就是"两个东西捆一份"）。第二，SkipList 加了 clear() 和禁止拷贝两行——这来自一次真实的段错误（Segmentation Fault，访问了非法内存导致程序崩溃）：我本来想用 memtable_ = SkipList() 换一个新表，但 SkipList 里只有裸指针 head_，C++ 默认的拷贝赋值只复制指针本身不复制内容，等号右边的临时对象析构时把 head_ 指向的内存 delete 了，左边对象手里的 head_ 立刻变成悬空指针，一碰就崩。C++ 的三法则（Rule of Three）：类自己管了 new 出来的资源，就必须自己写析构函数、拷贝构造、拷贝赋值——或者像我们这样直接 = delete 把拷贝禁掉，改为提供 clear()。这个坑你早晚会踩，这里踩掉最好。
第三，MiniLSM 的 get 里多了三条统计（mem_hits_、sst_hits_、bloom_blocked_），毕业成绩单全靠它们。注意 mem_hits_ 在查到墓碑时也计数——删除的 key 走的也是 MemTable 路径。
编译运行：
g++ -std=c++17 -O2 mini_lsm.cpp -o mini_lsm
rm -f wal.log sst_*.data
./mini_lsm你应该看到（毫秒数在你机器上会不同）：
[flush] 写成 sst_1.data
[flush] 写成 sst_2.data
[flush] 写成 sst_3.data
[flush] 写成 sst_4.data
[compact] 有 4 个 SSTable，开始合并
[compact] 合并完成: 2000 条有效数据 -> sst_5.data
========== 毕业成绩单 ==========
写入 2000 条 + 删除 200 条 耗时: 145 毫秒
查询 500 次耗时: 36 毫秒
活着的 key 找到: 300/300
删掉的 key 正确返回不存在: 100/100
没存在的 key 正确返回不存在: 100/100
路径统计: MemTable 命中 100 次, SSTable 命中 300 次
布隆过滤器拦下磁盘白跑: 100 次对照 M1 体会一下：同样是 2000 条数据、500 次查询，M1 平均每次要扫接近全文件；现在 MemTable 命中 100 次（删除的 key，墓碑在内存里直接判死）、SSTable 命中 300 次（布隆放行 + 索引定位 + 小范围扫描）、100 个幽灵 key 全被布隆拦下。每条查询路径都清晰可控，没有任何一次全文件扫描——这就是 LSM 引擎存在的意义。
调试现场三：我第一版在 flush 里写 memtable_ = SkipList() 想换个新表，程序跑到一半直接段错误（Segfault）。用报错信息定位到崩溃点在跳表的 put 里，一看 head_ 是野指针——就是上面讲的三法则坑。修法是提供 clear() 并禁用拷贝。你写到这里如果也崩了，恭喜你踩到和我同一个坑，按同样方式修。
练习 把 MEMTABLE_LIMIT 调到 100，看 flush 和 compact 的节奏怎么变；再把 500 次查询改成 5000 次，看查询总耗时是不是大约线性增长。
M8第九课 M8：给引擎装上 Redis 的嘴——RESP 协议服务器
目标 一个真正的网络服务器。Redis 客户端连上它的 7379 端口，SET/GET/DEL/EXPIRE 全部能用。对应简历词条：Redis RESP 协议兼容。
先懂协议再写代码。RESP（REdis Serialization Protocol，Redis 序列化协议）是 Redis 客户端和服务器说话的格式。你敲 SET foo bar，客户端实际发到网络上的是一串带"货物清单"的文本：第一行 *3 表示"这条命令有 3 个参数"，然后每个参数前面用 $长度 报出自己的字节数，内容跟在后面，每行都以 \r\n（回车+换行）结尾：
*3
$3
SET
$3
foo
$3
bar服务器回应也讲格式：+OK 表示成功，$18\r\nxiaoming_score_135\r\n 表示"返回 18 字节的内容"，$-1 表示不存在，:1 返回整数 1，-ERR 开头是错误。你要做的只是：从网络字节流里按这套规矩"拆包"，执行命令，再按规矩"装包"回去。
新建 m8_redis_server/server.cpp。网络部分 Windows 和 Linux 的函数名不一样，用 #ifdef _WIN32 两边都写上（Linux 下编译不用加 -lws2_32）：
#include <iostream>
#include <fstream>
#include <map>
#include <string>
#include <vector>
#include <sstream>
#include <cstdlib>
#include <ctime>
// ============ 跨平台 socket 垫片 ============
#ifdef _WIN32
#include <winsock2.h>
#include <ws2tcpip.h>
#else
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <arpa/inet.h>
typedef int SOCKET;
#define INVALID_SOCKET (-1)
#define closesocket close
#endif
const std::string TOMBSTONE = "__TOMBSTONE__";
// ============ 极简 KV 引擎（WAL + MemTable + 过期时间）============
// M7 的完整引擎可以原样替换进来，服务器这层不用改
class MiniKV {
public:
MiniKV() {
std::ifstream in("wal.log");
std::string k, v;
while (in >> k >> v) memtable_[k] = v;
wal_.open("wal.log", std::ios::app);
}
void put(const std::string& key, const std::string& value) {
wal_ << key << " " << value << "\n";
wal_.flush();
memtable_[key] = value;
expire_.erase(key); // 新值覆盖旧值时，旧的过期时间作废
}
void del(const std::string& key) {
wal_ << key << " " << TOMBSTONE << "\n";
wal_.flush();
memtable_[key] = TOMBSTONE;
expire_.erase(key);
}
bool get(const std::string& key, std::string& value) {
is_expired(key); // 先顺手清理过期 key
std::map<std::string, std::string>::iterator it = memtable_.find(key);
if (it == memtable_.end()) return false;
if (it->second == TOMBSTONE) return false;
value = it->second;
return true;
}
// EXPIRE key 秒数：给 key 设一个"保质期"
int expire(const std::string& key, long long seconds) {
is_expired(key);
if (!memtable_.count(key) || memtable_[key] == TOMBSTONE) return 0;
expire_[key] = (long long)time(0) + seconds;
return 1;
}
private:
void is_expired(const std::string& key) {
std::map<std::string, long long>::iterator it = expire_.find(key);
if (it == expire_.end()) return;
if ((long long)time(0) >= it->second) {
memtable_[key] = TOMBSTONE; // 过期 = 悄悄写墓碑
expire_.erase(it);
}
}
std::map<std::string, std::string> memtable_;
std::map<std::string, long long> expire_; // key -> 过期时刻（Unix 秒）
std::ofstream wal_;
};
// ============ RESP 协议解析（Redis 序列化协议）============
// 逐字节读，直到 \n（顺手去掉前面的 \r）。返回 false 表示连接断了
bool recv_line(SOCKET fd, std::string& out) {
out.clear();
char c;
while (true) {
int n = recv(fd, &c, 1, 0);
if (n <= 0) return false;
if (c == '\n') {
if (!out.empty() && out[out.size() - 1] == '\r') out.erase(out.size() - 1);
return true;
}
out += c;
}
}
// 读一个"批量字符串"：$长度\r\n 内容 \r\n
bool recv_bulk(SOCKET fd, std::string& out) {
std::string line;
if (!recv_line(fd, line)) return false;
if (line.empty() || line[0] != '$') return false;
int len = atoi(line.c_str() + 1);
out.resize(len);
int got = 0;
while (got < len) {
int n = recv(fd, &out[0] + got, len - got, 0);
if (n <= 0) return false;
got += n;
}
std::string tail;
recv_line(fd, tail); // 吃掉内容后面跟着的 \r\n
return true;
}
// 读一条完整命令，拆成 ["SET", "foo", "bar"]
std::vector<std::string> recv_command(SOCKET fd) {
std::string line;
if (!recv_line(fd, line)) return std::vector<std::string>();
std::vector<std::string> args;
if (!line.empty() && line[0] == '*') {
int n = atoi(line.c_str() + 1);
for (int i = 0; i < n; i++) {
std::string arg;
if (!recv_bulk(fd, arg)) return std::vector<std::string>();
args.push_back(arg);
}
} else {
// 兼容"裸命令"：直接敲 SET foo bar 回车（telnet 测试用）
std::istringstream iss(line);
std::string w;
while (iss >> w) args.push_back(w);
}
return args;
}
// 发送应答
void send_str(SOCKET fd, const std::string& s) {
send(fd, s.c_str(), (int)s.size(), 0);
}
int main() {
#ifdef _WIN32
WSADATA wsa;
WSAStartup(MAKEWORD(2, 2), &wsa);
#endif
MiniKV db;
SOCKET listener = socket(AF_INET, SOCK_STREAM, 0);
if (listener == INVALID_SOCKET) {
std::cout << "socket 创建失败\n";
return 1;
}
int opt = 1;
setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
sockaddr_in addr;
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = INADDR_ANY; // 监听本机所有网卡
addr.sin_port = htons(7379); // 用 7379，避开真 Redis 的 6379
if (bind(listener, (sockaddr*)&addr, sizeof(addr)) < 0) {
std::cout << "bind 失败（端口被占？）\n";
return 1;
}
listen(listener, 5);
std::cout << "MiniRedis 已启动，监听 127.0.0.1:7379，等客户端来连...\n";
while (true) {
// accept：一直等到有客户端连进来才往下走
SOCKET conn = accept(listener, 0, 0);
if (conn == INVALID_SOCKET) continue;
std::cout << "[连接] 有客户端接入\n";
while (true) {
std::vector<std::string> args = recv_command(conn);
if (args.empty()) break; // 连接断了或命令解析失败
std::string cmd = args[0];
// 统一转大写，set/Set/SET 一视同仁
for (size_t i = 0; i < cmd.size(); i++) {
if (cmd[i] >= 'a' && cmd[i] <= 'z') cmd[i] = cmd[i] - 'a' + 'A';
}
if (cmd == "PING") {
send_str(conn, "+PONG\r\n");
} else if (cmd == "SET" && args.size() >= 3) {
db.put(args[1], args[2]);
send_str(conn, "+OK\r\n");
} else if (cmd == "GET" && args.size() >= 2) {
std::string v;
if (db.get(args[1], v)) {
std::ostringstream oss;
oss << "$" << v.size() << "\r\n" << v << "\r\n";
send_str(conn, oss.str());
} else {
send_str(conn, "$-1\r\n"); // Redis 规定：不存在用 $-1 表示
}
} else if (cmd == "DEL" && args.size() >= 2) {
std::string v;
bool existed = db.get(args[1], v);
db.del(args[1]);
std::ostringstream oss;
oss << ":" << (existed ? 1 : 0) << "\r\n"; // :1 删了，:0 本来就没有
send_str(conn, oss.str());
} else if (cmd == "EXPIRE" && args.size() >= 3) {
int n = db.expire(args[1], atoi(args[2].c_str()));
std::ostringstream oss;
oss << ":" << n << "\r\n";
send_str(conn, oss.str());
} else {
send_str(conn, "-ERR unknown command\r\n");
}
}
closesocket(conn);
std::cout << "[断开] 客户端走了\n";
}
#ifdef _WIN32
WSACleanup();
#endif
return 0;
}网络部分几个新名词。socket（套接字）可以理解成"电话机"：socket() 买一部电话，bind() 给它装上门牌号（IP 地址 + 端口号，端口就是一栋楼里的房间号，我们用 7379 避开真 Redis 的 6379），listen() 开始守着电话，accept() 等到有人打进来就接通，之后 recv()/send() 就是在这条线上收发字节。EXPIRE 的实现是"惰性过期"：给每个 key 记一个到期时刻，get 时顺手检查，到期就悄悄写墓碑——真实 Redis 也是主要靠惰性删除加定期抽样。
再准备一个测试客户端（m8_redis_server/test_client.py），用 Python 假装是 Redis 客户端：
# 用 Python 当 Redis 客户端，测试我们的 MiniRedis
# 用法: 先启动 server，再运行 python3 test_client.py
import socket
import time
s = socket.create_connection(("127.0.0.1", 7379), timeout=5)
def cmd(*args):
"""把命令拼成 RESP 格式发出去，等应答"""
msg = ("*" + str(len(args)) + "\r\n").encode()
for a in args:
b = str(a).encode()
msg += ("$" + str(len(b)) + "\r\n").encode() + b + b"\r\n"
s.sendall(msg)
time.sleep(0.1) # 给服务器一点反应时间
return s.recv(4096)
print("PING ->", cmd("PING"))
print("SET user:51 ->", cmd("SET", "user:51", "xiaoming_score_135"))
print("GET user:51 ->", cmd("GET", "user:51"))
print("GET nope ->", cmd("GET", "nope"))
print("DEL user:51 ->", cmd("DEL", "user:51"))
print("GET user:51 ->", cmd("GET", "user:51"))
print("SET user:52 ->", cmd("SET", "user:52", "alice"))
print("EXPIRE user:52->", cmd("EXPIRE", "user:52", "1"))
print("GET user:52 ->", cmd("GET", "user:52"), "(还没过期)")
time.sleep(2)
print("2秒后再 GET ->", cmd("GET", "user:52"), "(已过期)")
s.close()编译并测试（开两个终端，或者用 & 让服务器在后台跑；Windows 的 g++ 要加 -lws2_32，Linux 不用）：
g++ -std=c++17 -O2 server.cpp -o server -lws2_32
./server &
python3 test_client.py你应该看到：
PING -> b'+PONG\r\n'
SET user:51 -> b'+OK\r\n'
GET user:51 -> b'$18\r\nxiaoming_score_135\r\n'
GET nope -> b'$-1\r\n'
DEL user:51 -> b':1\r\n'
GET user:51 -> b'$-1\r\n'
SET user:52 -> b'+OK\r\n'
EXPIRE user:52-> b':1\r\n'
GET user:52 -> b'$5\r\nalice\r\n' (还没过期)
2秒后再 GET -> b'$-1\r\n' (已过期)你的存储引擎现在会通过网络说话了。测完别忘 kill %1（Linux）或 Ctrl+C 关掉服务器。如果你装了真的 redis-cli，也可以用 redis-cli -p 7379 连上来敲命令，效果一样——因为协议是真的。
调试 bind 失败多半是端口被占，换个端口或杀掉旧进程；Windows 忘加 -lws2_32 会报一堆 undefined reference to WSAStartup；客户端连不上先检查服务器是不是真的在跑（./server 有没有打印启动信息）。
练习 加一条 TTL 命令（返回 key 还剩几秒过期，没有则返回 -1）；把服务器改成支持多个客户端轮流连接（现在的 accept 循环天然支持"一个走了下一个再进"，想想为什么；真正的并发要用 epoll，这是进阶课）。
第十部分：毕业总结——简历、差距与下一步
你的代码和简历词条的对应关系
自开发 LSM-Tree 存储引擎及 Redis 协议服务器（C++）——这句话拆开，每个词你现在都有了亲手写过的对应物。设计并实现基于 LSM-Tree 的存储引擎：M2 到 M7 五课。写前日志 WAL 崩溃恢复：M2，断电实验亲手做过。跳表 MemTable：M3，查找路线都打印过。多级 SSTable 加稀疏索引：M4，seekg 定位亲手写的。Bloom Filter：M5，拦截次数亲眼看的。Compaction 策略与空间回收：M6，墓碑封杀旧值亲手修的。兼容 Redis RESP 协议：M8，GET/SET/DEL/EXPIRE 全部测通。性能调优与压测数据：M1 和 M7 的计时实验，把 M7 的规模调到 10 万条、用 chrono 记 P99 延迟，就是你简历上的数字来源——数字必须来自你机器上真实跑出的结果。
迷你版和真实引擎的差距地图
多级 SSTable：我们只有"最新的放最前"一个列表，LevelDB/RocksDB 分 L0 到 L6 层，L0 允许重叠、L1 以下每层 key 范围互不重叠，逐层向下合并，读放大更小。Leveled Compaction：我们是"超过 3 个全合并"，真实引擎按每层大小比例选文件向下合并，还要限流防止合并拖垮前台写入。跳表并发：我们的跳表单线程用，真实引擎用无锁或细粒度锁的并发跳表。epoll 事件循环：我们一次只服务一个连接，Redis 用单线程 epoll 同时服务几万个连接——这是你下一个值得啃的主题。编码细节：真实 SSTable 用 varint 压缩、CRC 校验、数据块加索引块加元数据块（footer），我们用最朴素的文本行。mmap、IO_uring：进阶的 I/O 手段，简历上出现这些词之前，先把它们各写一个 demo。
面试话术
被问"你做过存储相关的东西吗"，你可以这样说：我自己用 C++ 写过一个约六百行的 LSM-Tree KV 引擎和一个兼容 RESP 协议的 Redis 服务器。写入路径是先写 WAL 再进跳表 MemTable，MemTable 满了按序落盘成带稀疏索引的 SSTable，读路径先 MemTable 再从新到旧查 SSTable，每个 SSTable 前面有布隆过滤器挡不存在 key 的磁盘访问，SSTable 超过阈值触发合并，用墓碑机制处理删除。我踩过两个印象深刻的坑：一个是 Windows 文本模式把换行写成两个字节导致偏移量全错位，让我明白存储引擎为什么必须用二进制 I/O；一个是文件缓冲区没刷就去读，让我对"数据落盘"四个字有了敬畏——这两个坑比十篇博客记得牢。这段话里每个词你都亲手摸过，面试官往下追问任何一层你都能接住。