第 4 章 · 基本操作

4.1 打开与关闭数据库

基本打开方式

#include "leveldb/db.h"

leveldb::DB* db;
leveldb::Options options;

// 关键选项
options.create_if_missing = true;    // 不存在则创建
options.error_if_exists = false;     // 已存在不报错

leveldb::Status status = leveldb::DB::Open(options, "/path/to/db", &db);
if (!status.ok()) {
    std::cerr << "Open failed: " << status.ToString() << std::endl;
    // 处理错误...
}

// 使用完毕后关闭
delete db;

Options 详解

参数	类型	默认值	说明
`create_if_missing`	bool	false	数据库目录不存在时自动创建
`error_if_exists`	bool	false	数据库目录已存在时返回错误
`paranoid_checks`	bool	false	打开时进行严格数据校验
`write_buffer_size`	size_t	4MB	MemTable 大小上限
`max_open_files`	int	1000	最大打开文件数
`block_cache`	Cache*	8MB	数据块缓存
`block_size`	size_t	4096	SSTable 数据块大小
`compression`	CompressionType	kSnappyCompression	压缩类型
`filter_policy`	FilterPolicy*	nullptr	Bloom Filter 策略
`comparator`	Comparator*	BytewiseComparator	Key 比较器

ReadOptions 详解

参数	类型	默认值	说明
`verify_checksums`	bool	false	读取时校验 CRC
`fill_cache`	bool	true	读取的数据是否放入 Block Cache
`snapshot`	Snapshot*	nullptr	指定快照读取
`read_tier`	ReadTier	kReadAllTier	是否只读 memtable

WriteOptions 详解

参数	类型	默认值	说明
`sync`	bool	false	写入后是否 fsync 到磁盘

错误处理最佳实践

// LevelDB 的 Status 对象封装了错误信息
leveldb::Status s = db->Put(...);
if (!s.ok()) {
    if (s.IsNotFound()) {
        // Get 操作：Key 不存在
    } else if (s.IsCorruption()) {
        // 数据损坏
    } else if (s.IsIOError()) {
        // I/O 错误（磁盘满、权限等）
    } else if (s.IsInvalidArgument()) {
        // 参数错误
    }
    std::cerr << "Error: " << s.ToString() << std::endl;
}

4.2 Put — 写入数据

基本写入

// 基本用法
leveldb::Status s = db->Put(
    leveldb::WriteOptions(),    // 写选项
    "mykey",                     // Key
    "myvalue"                    // Value
);
assert(s.ok());

// 同步写入（保证持久化到磁盘）
leveldb::WriteOptions sync_opts;
sync_opts.sync = true;
s = db->Put(sync_opts, "critical_key", "important_data");

Key/Value 的类型

LevelDB 的 Key 和 Value 都是 leveldb::Slice，本质上是字节序列：

// 方式 1：字符串字面量
db->Put(wopts, "key1", "value1");

// 方式 2：std::string
std::string key = "user:1001";
std::string value = "张三";
db->Put(wopts, key, value);

// 方式 3：二进制数据
char binary_key[8];
memcpy(binary_key, &timestamp, 8);
db->Put(wopts, leveldb::Slice(binary_key, 8), binary_value);

实用示例：存储 JSON

#include <sstream>
#include "leveldb/db.h"

// 存储用户信息
void SaveUser(leveldb::DB* db, int user_id,
              const std::string& name, int age) {
    // Key: 带前缀的用户 ID
    std::string key = "user:" + std::to_string(user_id);

    // Value: 简单 JSON
    std::ostringstream json;
    json << "{\"name\":\"" << name
         << "\",\"age\":" << age << "}";

    leveldb::Status s = db->Put(leveldb::WriteOptions(), key, json.str());
    if (!s.ok()) {
        throw std::runtime_error("Put failed: " + s.ToString());
    }
}

// 读取用户信息
std::string GetUser(leveldb::DB* db, int user_id) {
    std::string key = "user:" + std::to_string(user_id);
    std::string value;
    leveldb::Status s = db->Get(leveldb::ReadOptions(), key, &value);
    if (s.IsNotFound()) {
        return "";  // 用户不存在
    }
    if (!s.ok()) {
        throw std::runtime_error("Get failed: " + s.ToString());
    }
    return value;
}

Slice 的生命周期

// ⚠️ 危险：Slice 不拥有数据！
leveldb::Slice key;
{
    std::string temp = "temporary_key";
    key = temp;
}  // temp 被销毁，key 变成悬空指针！

// ✅ 正确做法：确保 Slice 的底层数据在使用期间有效
std::string key_str = "safe_key";
db->Put(wopts, key_str, value);  // 直接传 std::string，隐式转换为 Slice

4.3 Get — 读取数据

基本读取

std::string value;
leveldb::Status s = db->Get(
    leveldb::ReadOptions(),    // 读选项
    "mykey",                    // Key
    &value                      // 输出参数
);

if (s.ok()) {
    std::cout << "Value: " << value << std::endl;
} else if (s.IsNotFound()) {
    std::cout << "Key not found" << std::endl;
} else {
    std::cerr << "Error: " << s.ToString() << std::endl;
}

带校验的读取

leveldb::ReadOptions ropts;
ropts.verify_checksums = true;  // 启用 CRC 校验

std::string value;
leveldb::Status s = db->Get(ropts, "important_key", &value);

读取性能优化

// 场景：大量顺序读取不需要缓存
leveldb::ReadOptions ropts;
ropts.fill_cache = false;  // 不污染 Block Cache

std::string value;
db->Get(ropts, "cold_key", &value);

💡 提示：fill_cache = false 适用于批量扫描或数据迁移场景，避免把热数据挤出缓存。

4.4 Delete — 删除数据

基本删除

leveldb::Status s = db->Delete(leveldb::WriteOptions(), "mykey");
if (!s.ok()) {
    // 注意：Delete 一个不存在的 Key 也返回 OK
    // 因为 LSM-Tree 中删除只是写入一个 "删除标记"（Tombstone）
}

LSM-Tree 删除的本质

Delete("user:1001") 的实际行为：

1. 向 MemTable 写入一条记录：
   Key: "user:1001"
   Sequence: (最新)
   Type: kTypeDeletion (删除标记 / Tombstone)

2. 之前的 "user:1001" 值仍然存在于 SSTable 中
3. 在 Compaction 过程中，Tombstone 才会真正清除旧数据

⚠️ 注意：
Delete 后立即 Get 可能返回 NotFound（从 MemTable 的 Tombstone 得知）
但底层数据并未立即删除，需要等待 Compaction 清理
大量删除会导致空间放大（Space Amplification）

删除后重新插入

db->Put(wopts, "key1", "value1");     // 写入
db->Delete(wopts, "key1");             // 删除
db->Put(wopts, "key1", "value2");     // 重新写入

std::string value;
db->Get(ropts, "key1", &value);       // 返回 "value2"

4.5 Iterator — 遍历数据

创建迭代器

leveldb::Iterator* it = db->NewIterator(leveldb::ReadOptions());

// 使用完毕必须删除
delete it;

基本遍历

leveldb::Iterator* it = db->NewIterator(leveldb::ReadOptions());

// 正向遍历所有数据
for (it->SeekToFirst(); it->Valid(); it->Next()) {
    std::cout << it->key().ToString() << " = "
              << it->value().ToString() << std::endl;
}

// 检查遍历是否出错
if (!it->status().ok()) {
    std::cerr << "Iterator error: " << it->status().ToString() << std::endl;
}

delete it;

反向遍历

leveldb::Iterator* it = db->NewIterator(leveldb::ReadOptions());

for (it->SeekToLast(); it->Valid(); it->Prev()) {
    std::cout << it->key().ToString() << " = "
              << it->value().ToString() << std::endl;
}

delete it;

Seek 到指定位置

leveldb::Iterator* it = db->NewIterator(leveldb::ReadOptions());

// Seek 到第一个 >= "user:100" 的 Key
it->Seek("user:100");
if (it->Valid()) {
    std::cout << "Found: " << it->key().ToString() << std::endl;
}

delete it;

前缀扫描

// 查找所有以 "user:" 开头的键
leveldb::Iterator* it = db->NewIterator(leveldb::ReadOptions());

std::string prefix = "user:";
for (it->Seek(prefix);
     it->Valid() && it->key().starts_with(prefix);
     it->Next()) {
    std::cout << it->key().ToString() << " = "
              << it->value().ToString() << std::endl;
}

delete it;

范围扫描

// 查找 Key 在 [start, end) 范围内的数据
leveldb::Iterator* it = db->NewIterator(leveldb::ReadOptions());

std::string start = "user:1000";
std::string end = "user:2000";

for (it->Seek(start); it->Valid(); it->Next()) {
    if (it->key().compare(end) >= 0) {
        break;  // 超出范围
    }
    std::cout << it->key().ToString() << " = "
              << it->value().ToString() << std::endl;
}

delete it;

只取前 N 条

// 取最新的 10 条记录
leveldb::Iterator* it = db->NewIterator(leveldb::ReadOptions());

int count = 0;
it->SeekToLast();
while (it->Valid() && count < 10) {
    std::cout << it->key().ToString() << std::endl;
    it->Prev();
    count++;
}

delete it;

安全的迭代器使用模式

// RAII 封装，确保迭代器被正确释放
class ScopedIterator {
public:
    explicit ScopedIterator(leveldb::DB* db,
                            const leveldb::ReadOptions& opts = leveldb::ReadOptions())
        : iter_(db->NewIterator(opts)) {}

    ~ScopedIterator() { delete iter_; }

    leveldb::Iterator* operator->() { return iter_; }
    leveldb::Iterator* get() { return iter_; }

    // 禁止复制
    ScopedIterator(const ScopedIterator&) = delete;
    ScopedIterator& operator=(const ScopedIterator&) = delete;

private:
    leveldb::Iterator* iter_;
};

// 使用
void ScanUsers(leveldb::DB* db) {
    ScopedIterator it(db);
    for (it->Seek("user:"); it->Valid(); it->Next()) {
        // ...
    }  // 自动 delete iter_
}

4.6 状态码参考

Status 方法	含义	常见场景
`ok()`	成功	正常操作
`IsNotFound()`	Key 不存在	Get 操作未找到
`IsCorruption()`	数据损坏	SSTable CRC 校验失败
`IsIOError()`	I/O 错误	磁盘满、文件不存在
`IsInvalidArgument()`	参数无效	错误的选项值
`IsNotSupported()`	不支持	读取已删除的迭代器
`IsTimedOut()`	超时	等待锁超时

4.7 完整实战：通讯录程序

#include <iostream>
#include <string>
#include <sstream>
#include "leveldb/db.h"

class ContactBook {
public:
    ContactBook(const std::string& path) {
        leveldb::Options opts;
        opts.create_if_missing = true;
        opts.write_buffer_size = 2 * 1024 * 1024;  // 2MB
        leveldb::Status s = leveldb::DB::Open(opts, path, &db_);
        if (!s.ok()) throw std::runtime_error(s.ToString());
    }
    ~ContactBook() { delete db_; }

    // 添加联系人
    bool Add(const std::string& name, const std::string& phone,
             const std::string& email) {
        std::string key = "contact:" + name;
        std::ostringstream val;
        val << phone << "|" << email;
        return db_->Put(leveldb::WriteOptions(), key, val.str()).ok();
    }

    // 查找联系人
    bool Find(const std::string& name, std::string& phone,
              std::string& email) {
        std::string value;
        leveldb::Status s = db_->Get(leveldb::ReadOptions(),
                                     "contact:" + name, &value);
        if (!s.ok()) return false;
        auto pos = value.find('|');
        phone = value.substr(0, pos);
        email = value.substr(pos + 1);
        return true;
    }

    // 删除联系人
    bool Remove(const std::string& name) {
        return db_->Delete(leveldb::WriteOptions(),
                           "contact:" + name).ok();
    }

    // 列出所有联系人
    void List() {
        leveldb::Iterator* it = db_->NewIterator(leveldb::ReadOptions());
        for (it->Seek("contact:"); it->Valid(); it->Next()) {
            if (!it->key().starts_with("contact:")) break;
            std::string name = it->key().ToString().substr(8);
            std::string val = it->value().ToString();
            auto pos = val.find('|');
            std::cout << "  " << name
                      << " | " << val.substr(0, pos)
                      << " | " << val.substr(pos + 1) << "\n";
        }
        delete it;
    }

    // 搜索（按名字前缀）
    void Search(const std::string& prefix) {
        leveldb::Iterator* it = db_->NewIterator(leveldb::ReadOptions());
        std::string seek_key = "contact:" + prefix;
        for (it->Seek(seek_key); it->Valid(); it->Next()) {
            if (!it->key().starts_with("contact:")) break;
            std::string name = it->key().ToString().substr(8);
            if (name.substr(0, prefix.size()) != prefix) break;
            std::string val = it->value().ToString();
            auto pos = val.find('|');
            std::cout << "  " << name
                      << " | " << val.substr(0, pos)
                      << " | " << val.substr(pos + 1) << "\n";
        }
        delete it;
    }

private:
    leveldb::DB* db_;
};

int main() {
    ContactBook book("/tmp/contacts");

    book.Add("张三", "138-0000-0001", "[email protected]");
    book.Add("张伟", "139-0000-0002", "[email protected]");
    book.Add("李四", "137-0000-0003", "[email protected]");
    book.Add("王五", "136-0000-0004", "[email protected]");

    std::cout << "=== 所有联系人 ===\n";
    book.List();

    std::cout << "\n=== 搜索 '张' ===\n";
    book.Search("张");

    std::cout << "\n=== 查找 '李四' ===\n";
    std::string phone, email;
    if (book.Find("李四", phone, email)) {
        std::cout << "  电话: " << phone << "\n";
        std::cout << "  邮箱: " << email << "\n";
    }

    return 0;
}

编译运行

g++ -std=c++17 -O2 -o contacts contacts.cpp -lleveldb -lpthread
./contacts

# 输出:
# === 所有联系人 ===
#   张三 | 138-0000-0001 | [email protected]
#   张伟 | 139-0000-0002 | [email protected]
#   李四 | 137-0000-0003 | [email protected]
#   王五 | 136-0000-0004 | [email protected]
#
# === 搜索 '张' ===
#   张三 | 138-0000-0001 | [email protected]
#   张伟 | 139-0000-0002 | [email protected]
#
# === 查找 '李四' ===
#   电话: 137-0000-0003
#   邮箱: [email protected]

4.8 常见错误与排查

问题	原因	解决方案
`IOError: .../LOCK: ...`	另一个进程已打开同一 DB	关闭其他进程或删除 LOCK 文件
`Corruption: ...`	SSTable 文件损坏	修复或从备份恢复
`Invalid argument: ...`	`error_if_exists=true` 且 DB 已存在	改用 `create_if_missing=true`
`Not found`	Key 不存在	检查 Key 是否正确，注意大小写
写入后读不到	未删除旧 Iterator 或 Snapshot	确保 Iterator/Snapshot 生命周期正确
耗尽文件描述符	`max_open_files` 太小	增大该值或检查 ulimit

4.9 本章小结

操作	核心 API	注意事项
打开	`DB::Open(options, path, &db)`	`create_if_missing` 设置
写入	`db->Put(wopts, key, value)`	Slice 不拥有内存
读取	`db->Get(ropts, key, &value)`	处理 `IsNotFound`
删除	`db->Delete(wopts, key)`	Tombstone 机制
遍历	`db->NewIterator(ropts)`	必须 `delete it`
关闭	`delete db`	RAII 封装推荐

扩展阅读

LevelDB API 参考：include/leveldb/db.h
Iterator 实现：db/db_iter.cc，理解内部合并迭代器
Slice 设计哲学：Google 的 StringPiece 模式

← 第 3 章 · 架构总览 | 第 5 章 · 批量写入 →

强曰为道

第 4 章 · 基本操作

第 4 章 · 基本操作

4.1 打开与关闭数据库

基本打开方式

Options 详解

ReadOptions 详解

WriteOptions 详解

错误处理最佳实践

4.2 Put — 写入数据

基本写入

Key/Value 的类型

实用示例：存储 JSON

Slice 的生命周期

4.3 Get — 读取数据

基本读取

带校验的读取

读取性能优化

4.4 Delete — 删除数据

基本删除

LSM-Tree 删除的本质

删除后重新插入

4.5 Iterator — 遍历数据

创建迭代器

基本遍历

反向遍历

Seek 到指定位置

前缀扫描

范围扫描

只取前 N 条

安全的迭代器使用模式

4.6 状态码参考

4.7 完整实战：通讯录程序

编译运行

4.8 常见错误与排查

4.9 本章小结

扩展阅读