从RocksDB到LevelDB：手把手教你用C++实现一个简易的LSM-Tree存储引擎

从RocksDB到LevelDB：手把手教你用C++实现一个简易的LSM-Tree存储引擎

在当今数据爆炸式增长的时代，高效可靠的存储引擎成为各类系统的核心组件。Google开源的LevelDB作为LSM-Tree(Log-Structured Merge-Tree)的经典实现，以其出色的写入性能和紧凑的设计吸引了大量开发者。而Facebook基于LevelDB开发的RocksDB，则通过引入多线程压缩、分层压缩等优化，进一步提升了性能表现。本文将带您从零开始，用C++实现一个具备核心功能的简易LSM-Tree存储引擎，深入理解其设计精髓。

1. LSM-Tree核心设计解析

LSM-Tree的核心思想是将随机写入转换为顺序写入，从而大幅提升IO性能。其架构通常包含以下几个关键组件：

• MemTable：内存中的有序数据结构，负责接收最新写入
• Immutable MemTable：只读的内存表，等待持久化到磁盘
• SSTable(Sorted String Table)：磁盘上的有序数据文件
• WAL(Write-Ahead Log)：预写日志，确保数据持久性

与传统B-Tree类存储引擎相比，LSM-Tree在写入性能上通常有数量级的提升。下表展示了主要性能对比：

实现一个基础LSM-Tree引擎需要解决几个关键问题：

1. 内存数据结构的选择与实现
2. SSTable文件格式设计
3. 压缩(Compaction)策略
4. 崩溃恢复机制

2. 内存表(MemTable)实现

MemTable作为LSM-Tree的内存组件，需要支持高效的插入和查询操作。LevelDB选择了跳表(SkipList)作为其实现，相比平衡树有以下优势：

template<typename Key, typename Value> class SkipList { private: struct Node { Key key; Value value; std::vector<Node*> forward; // 多层指针数组 Node(const Key& k, const Value& v, int level) : key(k), value(v), forward(level, nullptr) {} }; Node* head_; int max_level_; std::random_device rd_; std::mt19937 gen_; std::uniform_real_distribution<> dis_; public: SkipList() : max_level_(12), gen_(rd_()) { head_ = new Node(Key(), Value(), max_level_); } // 随机生成节点层数 int RandomLevel() { int level = 1; while (dis_(gen_) < 0.5 && level < max_level_) { level++; } return level; } void Insert(const Key& key, const Value& value) { std::vector<Node*> update(max_level_, nullptr); Node* current = head_; // 从最高层开始查找插入位置 for (int i = max_level_ - 1; i >= 0; --i) { while (current->forward[i] != nullptr && current->forward[i]->key < key) { current = current->forward[i]; } update[i] = current; } int level = RandomLevel(); Node* newNode = new Node(key, value, level); // 更新各层指针 for (int i = 0; i < level; ++i) { newNode->forward[i] = update[i]->forward[i]; update[i]->forward[i] = newNode; } } };

跳表的平均时间复杂度为O(log n)，虽然理论上不如平衡树的O(log n)稳定，但实现简单且并发友好。在实际应用中，还需要考虑线程安全问题，LevelDB通过在外部加锁的方式保证线程安全。

3. SSTable文件格式设计

SSTable是LSM-Tree在磁盘上的持久化存储形式，其核心特点是按键有序排列。一个完整的SSTable文件通常包含以下几个部分：

1. 数据块(Data Blocks)：存储实际的键值对
2. 元数据块(Meta Blocks)：如布隆过滤器等
3. 索引块(Index Block)：指向数据块的索引
4. 文件尾部(Footer)：指向元数据块和索引块

下面是一个简化的SSTable写入实现：

class SSTableBuilder { public: void Add(const Slice& key, const Slice& value) { if (block_.size() >= block_size_) { FlushBlock(); } // 记录前一个key的共享前缀长度，节省空间 int shared = 0; if (last_key_.size() > 0) { shared = PrefixMatch(last_key_, key); } int non_shared = key.size() - shared; // 写入共享前缀长度、非共享长度和value长度 block_.append(EncodeVarint32(shared)); block_.append(EncodeVarint32(non_shared)); block_.append(EncodeVarint32(value.size())); // 写入非共享key部分和value block_.append(key.data() + shared, non_shared); block_.append(value.data(), value.size()); last_key_.assign(key.data(), key.size()); } void Finish() { if (!block_.empty()) { FlushBlock(); } WriteIndex(); WriteFooter(); } private: void FlushBlock() { // 压缩并写入数据块 std::string compressed; Compress(block_, &compressed); uint64_t offset = writer_->Write(compressed); // 记录块位置信息用于构建索引 BlockHandle handle; handle.set_offset(offset); handle.set_size(compressed.size()); index_entries_.emplace_back(last_key_, handle); block_.clear(); } void WriteIndex() { // 构建并写入索引块 SSTableBuilder index_builder; for (const auto& entry : index_entries_) { index_builder.Add(entry.key, entry.handle.Encode()); } index_builder.Finish(); } };

注意：实际实现中需要考虑校验和(Checksum)、压缩选项等细节，这里做了适当简化。

4. 压缩(Compaction)策略实现

压缩是LSM-Tree的核心操作之一，主要解决以下问题：

• 清理已删除的数据
• 合并多个SSTable文件，减少文件数量
• 优化数据布局，提高查询效率

LevelDB采用分层压缩策略(Leveled Compaction)，将SSTable分为多个层级：

下面是一个简化的压缩过程实现：

class Compactor { public: Status Compact(Version* current, VersionEdit* edit) { // 选择需要压缩的文件 Compaction* c = PickCompaction(current); if (c == nullptr) return Status::OK(); // 执行压缩 Iterator* input = current->MakeInputIterator(c); Status s = DoCompactionWork(c, input); // 安装新版本 if (s.ok()) { InstallCompactionResults(c, edit); } delete c; return s; } private: Status DoCompactionWork(Compaction* c, Iterator* input) { std::string current_user_key; bool has_current_user_key = false; SequenceNumber last_sequence_for_key = kMaxSequenceNumber; while (input->Valid()) { Slice key = input->key(); // 检查是否需要停止压缩(手动触发停止) if (compact_->ShouldStop()) { break; } // 处理相同key的不同版本 if (has_current_user_key && user_comparator_->Compare(key, current_user_key) == 0) { // 对于相同key，只保留最新版本 if (last_sequence_for_key <= compact_->smallest_snapshot) { builder_->Add(key, input->value()); } last_sequence_for_key = input->sequence(); } else { current_user_key.assign(key.data(), key.size()); has_current_user_key = true; last_sequence_for_key = input->sequence(); if (last_sequence_for_key <= compact_->smallest_snapshot) { builder_->Add(key, input->value()); } } input->Next(); } return builder_->Finish(); } };

压缩策略的选择对性能影响很大，RocksDB在LevelDB基础上引入了以下优化：

• 并行压缩：利用多线程加速压缩过程
• 分层压缩：更精细地控制各层文件大小
• 压缩优先级：根据文件热度决定压缩顺序

5. 崩溃恢复与WAL实现

为确保数据安全，LSM-Tree采用预写日志(WAL)机制。每次写入操作会先记录到日志文件，再应用到MemTable。崩溃恢复时，通过重放日志文件恢复数据。

日志文件格式设计要点：

• 使用CRC32校验和检测数据损坏
• 记录类型标识(如首记录、正常记录等)
• 批量写入提高性能

下面是一个简化的日志写入实现：

class LogWriter { public: Status AddRecord(const Slice& record) { const char* ptr = record.data(); size_t left = record.size(); // 分片写入，每个片段有头部信息 bool begin = true; do { const int leftover = kBlockSize - block_offset_; if (leftover < kHeaderSize) { // 填充空位 if (leftover > 0) { dest_->append(kZeroBytes, leftover); } block_offset_ = 0; } const size_t avail = kBlockSize - block_offset_ - kHeaderSize; const size_t fragment_length = (left < avail) ? left : avail; RecordType type; const bool end = (left == fragment_length); if (begin && end) { type = kFullType; } else if (begin) { type = kFirstType; } else if (end) { type = kLastType; } else { type = kMiddleType; } Status s = EmitPhysicalRecord(type, ptr, fragment_length); if (!s.ok()) return s; ptr += fragment_length; left -= fragment_length; begin = false; } while (left > 0); return Status::OK(); } private: Status EmitPhysicalRecord(RecordType t, const char* ptr, size_t n) { // 构造记录头 char buf[kHeaderSize]; buf[4] = static_cast<char>(n & 0xff); buf[5] = static_cast<char>(n >> 8); buf[6] = static_cast<char>(t); // 计算CRC32校验和 uint32_t crc = crc32c::Value(ptr, n); crc = crc32c::Extend(crc, buf + 4, 3); // 扩展计算头部 EncodeFixed32(buf, crc32c::Mask(crc)); // 写入头部和记录内容 Status s = dest_->Append(Slice(buf, kHeaderSize)); if (s.ok()) { s = dest_->Append(Slice(ptr, n)); if (s.ok()) { s = dest_->Flush(); } } block_offset_ += kHeaderSize + n; return s; } };

提示：实际实现中需要考虑日志轮转、批量提交等优化手段，以进一步提升性能。

6. 性能优化技巧

在完成基础功能后，可以考虑以下优化手段提升性能：

1. 布隆过滤器：为每个SSTable添加布隆过滤器，加速查询

   class BloomFilter { public: void Add(const Slice& key) { uint32_t h = hash_func_(key); for (int i = 0; i < k_; ++i) { bits_[(h + i * h) % bits_.size()] = true; } } bool MayContain(const Slice& key) const { uint32_t h = hash_func_(key); for (int i = 0; i < k_; ++i) { if (!bits_[(h + i * h) % bits_.size()]) { return false; } } return true; } };

2. 块缓存：缓存热点数据块，减少IO

   class BlockCache { public: Slice* Get(const CacheKey& key) { auto it = cache_.find(key); if (it != cache_.end()) { // 更新LRU队列 lru_list_.erase(it->second.lru_pos); lru_list_.push_front(key); it->second.lru_pos = lru_list_.begin(); return &it->second.data; } return nullptr; } void Insert(const CacheKey& key, const Slice& value) { if (cache_.size() >= capacity_) { // 淘汰最久未使用的数据 CacheKey old_key = lru_list_.back(); lru_list_.pop_back(); cache_.erase(old_key); } lru_list_.push_front(key); cache_[key] = {value, lru_list_.begin()}; } };

3. 多线程压缩：利用多核CPU并行执行压缩任务

   class CompactionScheduler { public: void ScheduleCompaction(CompactionTask* task) { std::lock_guard<std::mutex> lock(mutex_); pending_tasks_.push(task); cond_.notify_one(); } void WorkerThread() { while (true) { CompactionTask* task = nullptr; { std::unique_lock<std::mutex> lock(mutex_); cond_.wait(lock, [this]{ return !pending_tasks_.empty() || shutdown_; }); if (shutdown_) break; task = pending_tasks_.front(); pending_tasks_.pop(); } ExecuteCompaction(task); } } };

4. 写入批处理：合并小写入，减少IO次数

   class WriteBatch { public: void Put(const Slice& key, const Slice& value) { rep_.push_back(static_cast<char>(kTypeValue)); PutLengthPrefixedSlice(&rep_, key); PutLengthPrefixedSlice(&rep_, value); count_++; } void Clear() { rep_.clear(); count_ = 0; } Status Execute(DBImpl* db) { Writer w(this); Status s = db->Write(&w); if (s.ok()) { s = w.status; } return s; } };

7. 测试与性能评估

完成实现后，需要设计全面的测试验证系统正确性和性能。主要测试类型包括：

1. 功能测试：

   • 基本CRUD操作
   • 迭代器功能
   • 快照隔离
   • 并发控制

2. 性能测试：

   • 随机写入吞吐量
   • 随机读取延迟
   • 顺序扫描性能
   • 压缩对性能的影响

3. 稳定性测试：

   • 长时间运行测试
   • 崩溃恢复测试
   • 边界条件测试(如大value、空key等)

下面是一个简单的性能测试示例：

void RunBenchmark(DB* db, int num_entries) { // 写入测试 auto start = std::chrono::high_resolution_clock::now(); for (int i = 0; i < num_entries; ++i) { std::string key = "key" + std::to_string(i); std::string value = "value" + std::to_string(i); db->Put(WriteOptions(), key, value); } auto end = std::chrono::high_resolution_clock::now(); std::chrono::duration<double> elapsed = end - start; std::cout << "Write throughput: " << num_entries / elapsed.count() << " ops/s\n"; // 读取测试 start = std::chrono::high_resolution_clock::now(); for (int i = 0; i < num_entries; ++i) { std::string key = "key" + std::to_string(i); std::string value; db->Get(ReadOptions(), key, &value); } end = std::chrono::high_resolution_clock::now(); elapsed = end - start; std::cout << "Read throughput: " << num_entries / elapsed.count() << " ops/s\n"; }

在实际项目中，建议使用更专业的基准测试工具如YCSB(雅虎云服务基准测试)进行更全面的评估。