[转]LevelDB原理探究与代码分析

忘是亡心i 2022-08-22 12:26 280阅读 0赞

[LevelDB原理探究与代码分析][LevelDB]

# 1. 概述 #

Level DB（ [http://code.google.com/p/leveldb/][http_code.google.com_p_leveldb]）是google开源的Key/Value存储系统，它的committer阵容相当强大，基本上是bigtable的原班人马，包括像jeff dean这样的大牛，它的代码合设计非常具有借鉴意义，是一种典型的LSM Tree的KV引擎的实现，从它的数据结构来看，基本就是sstable的开源实现，而且针对各种平台作了port，目前被用在chrome等项目中。

#     #

# 2. LSM Tree #

Level DB是典型的Log-Structured-Merge Tree的实现，它通过延迟写入以及Write Log Ahead技术来加速数据的写入并保障数据的安全。LevelDB的每个数据文件(sstable)中的记录都是按照Key的顺序进行排序的，但是随机写入时，key的到来是无序的，因此难以将记录插入到其排序位置。于是需要它采取一种延迟写入的方式，批量攒集一定量的数据，将它们在内存中排好序，一次性写入到磁盘中。但是这期间一旦系统断电或其他异常，则可能导致数据丢失，因此需要将数据先写入到log的文件中，这样便将随机写转化为追加写入，对于磁盘性能会有很大提升，如果进程发生中断，重启后可以根据log恢复之前写入的数据。

## 2.1 Write Batch ##

Level DB只支持两种更新操作:

1. 插入一条记录

2. 删除一条记录

代码如下：

**\[cpp\]**  [view plain][]   [copy][view plain]

1.  std::string key1,key2,value;    
2.  leveldb::Status s;  
3.  s = db->Put(leveldb::WriteOptions(), key1, value);    
4.  s = db->Delete(leveldb::WriteOptions(), key2);

同时还支持以一种批量的方式写入数据：

**\[cpp\]**  [view plain][]   [copy][view plain]

1.  std::string key1,key2,value;   
2.  leveldb::WriteBatch batch;   
3.  batch.Delete(key1);   
4.  batch.Put(key2, value);   
5.  leveldb::status s = db->Write(leveldb::WriteOptions(), &batch);

其实，在Level DB内部，单独更新与批量更新的调用的接口是相同的，单独更新也会被组织成为包含一条记录的Batch，然后写入数据库中。Write Batch的组织形式如下：

![1335447663_7043.png][]

## 2.1 Log Format ##

每次更新操作都被组织成这样一个数据包，并作为一条日志写入到log文件中，同时也会被解析为一条条内存记录，按照key排序后插入到内存表中的相应位置。LevelDB使用Memory Mapping的方式对log数据进行访问：如果前一次映射的空间已写满，则先将文件扩展一定的长度（每次扩展的长度按64KB,128KB,...的顺序逐次翻倍，最大到1MB），然后映射到内存，对映射的内存再以32KB的Page进行切分，每次写入的日志填充到Page中，攒积一定量后Sync到磁盘上（也可以设置WriteOptions，每写一条日志就Sync一次，但是这样效率很低），内存映射文件的代码如下：

**\[cpp\]**  [view plain][]   [copy][view plain]

1.  **class** PosixMmapFile : **public** WritableFile   
2.  \{  
3.  **private**:  
4.    std::string filename\_;  // 文件名称  
5.    **int** fd\_;                // 文件句柄  
6.    **size\_t** page\_size\_;      //   
7.    **size\_t** map\_size\_;       // 内存映射的区域大小  
8.    **char**\* base\_;            // 内存映射区域的起始地址  
9.    **char**\* limit\_;           // 内存映射区域的结束地址  
10.   **char**\* dst\_;             // 最后一次占用的内存的结束地址  
11.   **char**\* last\_sync\_;       // 最后一次同步到磁盘的结束地址  
12.   uint64\_t file\_offset\_;  // 当前文件的偏移值  
13.   **bool** pending\_sync\_;     // 延迟同步的标志  
14.    
15. **public**:  
16.   PosixMmapFile(**const** std::string& fname, **int** fd, **size\_t** page\_size)  
17.       : filename\_(fname),  
18.         fd\_(fd),  
19.         page\_size\_(page\_size),  
20.         map\_size\_(Roundup(65536, page\_size)),  
21.         base\_(NULL),  
22.         limit\_(NULL),  
23.         dst\_(NULL),  
24.         last\_sync\_(NULL),  
25.         file\_offset\_(0),  
26.         pending\_sync\_(**false**) \{  
27.     assert((page\_size & (page\_size - 1)) == 0);  
28.   \}  
29.   
30.   ~PosixMmapFile() \{  
31.     **if** (fd\_ >= 0) \{  
32.       PosixMmapFile::Close();  
33.     \}  
34.   \}  
35.   
36.   Status Append(**const** Slice& data) \{  
37.     **const** **char**\* src = data.data();  
38.     **size\_t** left = data.size();  
39.     **while** (left > 0) \{  
40.       // 计算上次最后一次申请的区域的剩余容量，如果已完全耗尽，  
41.       // 则卸载当前区域，申请一个新的区域  
42.       **size\_t** avail = limit\_ - dst\_;  
43.       **if** (avail == 0) \{  
44.         **if** (!UnmapCurrentRegion() ||  
45.             !MapNewRegion()) \{  
46.           **return** IOError(filename\_, errno);  
47.         \}  
48.       \}  
49.       // 填充当前区域的剩余容量  
50.       **size\_t** n = (left <= avail) ? left : avail;  
51.       memcpy(dst\_, src, n);  
52.       dst\_ += n;  
53.       src += n;  
54.       left -= n;  
55.     \}  
56.     **return** Status::OK();  
57.   \}  
58.   
59.   Status PosixMmapFile::Close() \{  
60.     Status s;  
61.     **size\_t** unused = limit\_ - dst\_;  
62.     **if** (!UnmapCurrentRegion()) \{  
63.       s = IOError(filename\_, errno);  
64.     \} **else** **if** (unused > 0) \{  
65.       // 关闭时将文件没有使用用的空间truncate掉  
66.       **if** (ftruncate(fd\_, file\_offset\_ - unused) < 0) \{  
67.         s = IOError(filename\_, errno);  
68.       \}  
69.     \}  
70.   
71.     **if** (close(fd\_) < 0) \{  
72.       **if** (s.ok()) \{  
73.         s = IOError(filename\_, errno);  
74.       \}  
75.     \}  
76.   
77.     fd\_ = -1;  
78.     base\_ = NULL;  
79.     limit\_ = NULL;  
80.     **return** s;  
81.   \}  
82.     
83.   **virtual** Status Sync() \{  
84.     Status s;  
85.     **if** (pending\_sync\_) \{  
86.       // 上个区域也有数据未同步，则先同步数据  
87.       pending\_sync\_ = **false**;  
88.       **if** (fdatasync(fd\_) < 0) \{  
89.         s = IOError(filename\_, errno);  
90.       \}  
91.     \}  
92.   
93.     **if** (dst\_ > last\_sync\_) \{  
94.       // 计算未同步数据的起始与结束地址，同步时，起始地址按page\_size\_向下取整，  
95.       // 结束地址向上取整，保证每次同步都是同步一个或多个page  
96.       **size\_t** p1 = TruncateToPageBoundary(last\_sync\_ - base\_);  
97.       **size\_t** p2 = TruncateToPageBoundary(dst\_ - base\_ - 1);   
98.       // 如果刚好为整数个page\_size\_，由于下面同步时必然会加一个page\_size\_，所以这里可以减去1  
99.       last\_sync\_ = dst\_;  
100.       **if** (msync(base\_ + p1, p2 - p1 + page\_size\_, MS\_SYNC) < 0) \{  
101.         s = IOError(filename\_, errno);  
102.       \}  
103.     \}  
104.     **return** s;  
105.   \}  
106. **private**:  
107.   // 将x按y向上对齐     
108.   **static** **size\_t** Roundup(**size\_t** x, **size\_t** y) \{  
109.     **return** ((x + y - 1) / y) \* y;  
110.   \}  
111.   // 将s按page\_size\_向下对齐  
112.   **size\_t** TruncateToPageBoundary(**size\_t** s) \{  
113.     s -= (s & (page\_size\_ - 1));  
114.     assert((s % page\_size\_) == 0);  
115.     **return** s;  
116.   \}   
117.   
118.   // 卸载当前映射的内存区域    
119.   **bool** UnmapCurrentRegion() \{      
120.     **bool** result = **true**;  
121.     **if** (base\_ != NULL) \{  
122.       **if** (last\_sync\_ < limit\_) \{  
123.         // 如果当前页没有完全被同步，则标明本文件需要被同步，下次调用Sync()方法时会将本页中未同步的数据同步到磁盘  
124.         pending\_sync\_ = **true**;  
125.       \}  
126.       **if** (munmap(base\_, limit\_ - base\_) != 0) \{  
127.         result = **false**;  
128.       \}  
129.       file\_offset\_ += limit\_ - base\_;  
130.       base\_ = NULL;  
131.       limit\_ = NULL;  
132.       last\_sync\_ = NULL;  
133.       dst\_ = NULL;      // 使用翻倍的策略增加下次申请区域的大小，最大到1MB  
134.       **if** (map\_size\_ < (1<<20)) \{  
135.         map\_size\_ \*= 2;  
136.       \}  
137.     \}  
138.     **return** result;  
139.   \}    
140.   
141.   **bool** MapNewRegion() \{  
142.     assert(base\_ == NULL); // 申请一个新的区域时，上一个申请的区域必须已经卸载   
143.     // 先将文件扩大      
144.     **if** (ftruncate(fd\_, file\_offset\_ + map\_size\_) < 0) \{  
145.       **return** **false**;  
146.     \}  
147.     // 将新区域映射到文件  
148.     **void**\* ptr = mmap(NULL, map\_size\_, PROT\_READ | PROT\_WRITE, MAP\_SHARED,  
149.                      fd\_, file\_offset\_);  
150.     **if** (ptr == MAP\_FAILED) \{  
151.       **return** **false**;  
152.     \}  
153.     base\_ = **reinterpret\_cast**<**char**\*>(ptr);  
154.     limit\_ = base\_ + map\_size\_;  
155.     dst\_ = base\_;  
156.     last\_sync\_ = base\_;  
157.     **return** **true**;  
158.   \}  
159. \};

但是，一个Batch的数据按上面的方式组织后，如果做为一条日志写入Log，则很可能需要跨两个或更多个Page；为了更好地管理日志以及保障数据安全，LevelDB对日志记录进行了更细的切分，如果一个Batch对应的数据需要跨页，则会将其切分为多条Entry，然后写入到不同Page中，Entry不会跨越Page，我们通过对多个Entry进行解包，可以还原出的Batch数据。最终，LevelDB的log文件被组织为下面的形式：

![1335447972_2102.png][]

这里，我们可以看一下log\_writer的代码：

**\[cpp\]**  [view plain][]   [copy][view plain]

1.  Status Writer::AddRecord(**const** Slice& slice) \{  
2.    **const** **char**\* ptr = slice.data();  
3.    **size\_t** left = slice.size();  
4.    
5.    Status s;  
6.    **bool** begin = **true**;  
7.    **do** \{  
8.      **const** **int** leftover = kBlockSize - block\_offset\_;  
9.      assert(leftover >= 0);  
10.     **if** (leftover < kHeaderSize) \{  
11.       // 如果当前page的剩余长度小于7字节且大于0，则都填充'\\0'，并新起一个page  
12.       **if** (leftover > 0) \{  
13.         assert(kHeaderSize == 7);  
14.         dest\_->Append(Slice("\\x00\\x00\\x00\\x00\\x00\\x00", leftover));  
15.       \}  
16.       block\_offset\_ = 0;  
17.     \}  
18.     
19.     // 计算page能否容纳整体日志，如果不能，则将日志切分为多条entry，插入不同的page中，type中注明该entry是日志的开头部分，中间部分还是结尾部分。  
20.     **const** **size\_t** avail = kBlockSize - block\_offset\_ - kHeaderSize;  
21.     **const** **size\_t** fragment\_length = (left < avail) ? left : avail;  
22.   
23.     RecordType type;  
24.     **const** **bool** end = (left == fragment\_length);  
25.     **if** (begin && end) \{  
26.       type = kFullType;   // 本Entry保存完整的Batch  
27.     \} **else** **if** (begin) \{  
28.       type = kFirstType;  // 本Entry只保存起始部分  
29.     \} **else** **if** (end) \{  
30.       type = kLastType;   // 本Entry只保存结束部分  
31.     \} **else** \{  
32.       type = kMiddleType; // 本Entry保存Batch的中间部分，不含起始与结尾，有时可能需要保存多个middle  
33.     \}  
34.   
35.     s = EmitPhysicalRecord(type, ptr, fragment\_length);  
36.     ptr += fragment\_length;  
37.     left -= fragment\_length;  
38.     begin = **false**;  
39.   \} **while** (s.ok() && left > 0);  
40.   **return** s;  
41. \}  
42.   
43. Status Writer::EmitPhysicalRecord(RecordType t, **const** **char**\* ptr, **size\_t** n) \{  
44.   assert(n <= 0xffff);    
45.   assert(block\_offset\_ + kHeaderSize + n <= kBlockSize);  
46.   
47.   // 填充记录头  
48.   **char** buf\[kHeaderSize\];  
49.   buf\[4\] = **static\_cast**<**char**>(n & 0xff);  
50.   buf\[5\] = **static\_cast**<**char**>(n >> 8);  
51.   buf\[6\] = **static\_cast**<**char**>(t);  
52.   
53.   // 计算crc  
54.   uint32\_t crc = crc32c::Extend(type\_crc\_\[t\], ptr, n);  
55.   crc = crc32c::Mask(crc);   
56.   EncodeFixed32(buf, crc);  
57.   
58.   // 填充entry内容  
59.   Status s = dest\_->Append(Slice(buf, kHeaderSize));  
60.   **if** (s.ok()) \{  
61.     s = dest\_->Append(Slice(ptr, n));  
62.     **if** (s.ok()) \{  
63.       s = dest\_->Flush();  
64.     \}  
65.   \}  
66.   block\_offset\_ += kHeaderSize + n;  
67.   **return** s;  
68. \}

## 2.3 Write Log Ahead ##

Level DB在更新时，先写log，然后更新memtable，每个memtable会设置一个最大容量，如果超过阈值，则采用双buffer机制，关闭当前log文件并将当前memtable切换未从memtable，然后新建一个log文件以及memtable，将数据写进新的log文件与memtable，并通知后台线程对从memtable进行处理，及时将其dump到磁盘上，或者启动compaction流程。Write的代码分析如下：

**\[cpp\]**  [view plain][]   [copy][view plain]

1.  Status DBImpl::Write(**const** WriteOptions& options, WriteBatch\* updates)   
2.  \{  
3.    Status status;  
4.    MutexLock l(&mutex\_);  // 锁定互斥体，同一时间只能有一个线程更新数据  
5.    LoggerId self;     
6.    // 获取Logger的使用权,如果有其他线程拥有所有权，则等待至其释放所有权。  
7.    AcquireLoggingResponsibility(&self);  
8.    status = MakeRoomForWrite(**false**);  // May temporarily release lock and wait  
9.    uint64\_t last\_sequence = versions\_->LastSequence();  // 获取当前的版本号  
10.   **if** (status.ok()) \{  
11.     // 将当前版本号加1后作为本次更新的日志的版本，  
12.     // 一次批量更新可能包含多个操作，这些操作都用一个版本有一个好处：  
13.     // 本次更新的所有操作，要么都可见，要么都不可见，不存在一部分可见，另一部分不可见的情况。  
14.     WriteBatchInternal::SetSequence(updates, last\_sequence + 1);  
15.     // 但是本次更新可能有多个操作，跳过与操作数相等的版本号，保证不被使用  
16.     last\_sequence += WriteBatchInternal::Count(updates);  
17.   
18.     // 将batch写入log，然后应用到memtable中  
19.     \{  
20.       assert(logger\_ == &self);  
21.       mutex\_.Unlock();  
22.       // 这里，可以解锁，因为在AcquireLoggingResponsibility()方法中已经获取了Logger的拥有权，  
23.       // 其他线程即使获得了锁，但是由于&self != logger，其会阻塞在AcquireLoggingResponsibility()方法中。  
24.       // 将更新写入log文件，如果设置了每次写入进行sync，则将其同步到磁盘，这个操作可能比较长，  
25.       // 防止了mutex\_对象长期被占用，因为其还负责其他一些资源的同步  
26.       status = log\_->AddRecord(WriteBatchInternal::Contents(updates));  
27.       **if** (status.ok() && options.sync) \{  
28.         status = logfile\_->Sync();  
29.       \}  
30.       **if** (status.ok()) \{  
31.         // 成功写入了log后，才写入memtable  
32.         status = WriteBatchInternal::InsertInto(updates, mem\_);  
33.       \}  
34.       // 重新锁定mutex\_  
35.       mutex\_.Lock();  
36.       assert(logger\_ == &self);  
37.     \}  
38.     // 更新版本号  
39.     versions\_->SetLastSequence(last\_sequence);  
40.   \}  
41.   // 释放对logger的所有权，并通知等待的线程，然后解锁  
42.   ReleaseLoggingResponsibility(&self);  
43.   **return** status;  
44. \}  
45.   
46. // force参数表示强制新起一个memtable  
47. Status DBImpl::MakeRoomForWrite(**bool** force) \{  
48.   mutex\_.AssertHeld();  
49.   assert(logger\_ != NULL);  
50.   **bool** allow\_delay = !force;  
51.   Status s;  
52.   **while** (**true**) \{  
53.     **if** (!bg\_error\_.ok()) \{  
54.       // 后台线程存在问题，则返回错误，不接受更新  
55.       s = bg\_error\_;  
56.       **break**;  
57.     \} **else** **if** (  
58.         allow\_delay &&  
59.         versions\_->NumLevelFiles(0) >= config::kL0\_SlowdownWritesTrigger) \{  
60.       // 如果不是强制写入，而且level 0的sstable超过8个，则本次更新阻塞1毫秒，  
61.       // leveldb将sstable分为多个等级，其中level 0中的不同表的key是可能重叠的，  
62.       // 如果l0的sstable过多，会导致查询性能下降，这时需要适当降低更新速度，让  
63.       // 后台线程进行compaction操作，但是设计者不希望让某次写操作等待数秒，  
64.       // 而是让每次更新操作分担延迟，即每次写操作阻塞1毫秒，平衡读写速率；  
65.       // 另外，理论上这也能让compaction线程获得更多的cpu时间（当然，  
66.       // 这是假定compaction与更新操作共享一个CPU时才有意义）  
67.       mutex\_.Unlock();  
68.       env\_->SleepForMicroseconds(1000);  
69.       allow\_delay = **false**;  // 最多延迟一次，下次不延迟  
70.       mutex\_.Lock();  
71.     \} **else** **if** (!force &&  
72.                (mem\_->ApproximateMemoryUsage() <= options\_.write\_buffer\_size)) \{  
73.       // 如果当前memtable已使用的空间小于write\_buffer\_size，则跳出，更新到当前memtable即可。  
74.       // 当force为true时，第一次循环会走后面else逻辑，切换了memtable后force被置为false，  
75.       // 第二次循环时就可以在此跳出了  
76.       **break**;  
77.     \} **else** **if** (imm\_ != NULL) \{  
78.       // 如果当前memtable已经超过write\_buffer\_size,且备用的memtable也在被使用，则阻塞更新并等待  
79.       bg\_cv\_.Wait();  
80.     \} **else** **if** (versions\_->NumLevelFiles(0) >= config::kL0\_StopWritesTrigger) \{  
81.       // 如果当前memtable已使用的空间小于write\_buffer\_size，但是备用的memtable未被使用，  
82.       // 则检查level 0的sstable个数，如超过12个，则阻塞更新并等待  
83.       Log(options\_.info\_log, "waiting...\\n");  
84.       bg\_cv\_.Wait();  
85.     \} **else** \{  
86.       // 否则，使用新的id新创建一个log文件，并将当前memtable切换为备用的memtable，新建一个  
87.       // memtable，然后将数据写入当前的新memtable，即切换log文件与memtable，并告诉后台线程  
88.       // 可以进行compaction操作了  
89.       assert(versions\_->PrevLogNumber() == 0);  
90.       uint64\_t new\_log\_number = versions\_->NewFileNumber();  
91.       WritableFile\* lfile = NULL;  
92.       s = env\_->NewWritableFile(LogFileName(dbname\_, new\_log\_number), &lfile);  
93.       **if** (!s.ok()) \{  
94.         **break**;  
95.       \}  
96.       **delete** log\_;  
97.       **delete** logfile\_;  
98.       logfile\_ = lfile;  
99.       logfile\_number\_ = new\_log\_number;  
100.       log\_ = **new** log::Writer(lfile);  
101.       imm\_ = mem\_;  
102.       has\_imm\_.Release\_Store(imm\_);  
103.       mem\_ = **new** MemTable(internal\_comparator\_);  
104.       mem\_->Ref();  
105.       force = **false**;   // 下次判断可以不新建memtable了  
106.       MaybeScheduleCompaction();  
107.     \}  
108.   \}  
109.   **return** s;  
110. \}  
111. **void** DBImpl::AcquireLoggingResponsibility(LoggerId\* self) \{  
112.   **while** (logger\_ != NULL) \{  
113.     logger\_cv\_.Wait();  
114.   \}  
115.   logger\_ = self;  
116. \}  
117.   
118. **void** DBImpl::ReleaseLoggingResponsibility(LoggerId\* self) \{  
119.   assert(logger\_ == self);  
120.   logger\_ = NULL;  
121.   logger\_cv\_.SignalAll();  
122. \}

## 2.4 Skip List    ##

Level DB内部采用跳表结构来组织Memtable，每插入一条记录，先根据跳表通过多次key的比较，定位到记录应该插入的位置，然后按照一定的概率确定该节点需要建立多少级的索引，跳表结构如下：

![1335523511_6056.png][]

Level DB的SkipList最高12层，最下面一层（level0）的链是全链，即每条记录必须在此链中插入相应的索引节点；从level1到level11则是按概率决定是否需要建索引，概率按照1/4的因子等比递减。下面举个例子，说明一下这个流程：  
1. 看上图，假定我们链不存在record3，level0中，record2的下一条记录是record4，level1中，record2的下一条记录是record5。  
2. 现在，我们插入一条记录record3，通过key的比较，我们定位到它应该在record2与record4之间。  
3. 然后，我们按照下面的代码确定一条记录需要在跳表中建立几重索引：

**\[cpp\]**  [view plain][]   [copy][view plain]

1.  **template**<**typename** Key, **class** Comparator>  
2.  **int** SkipList<Key,Comparator>::RandomHeight() \{  
3.    // Increase height with probability 1 in kBranching  
4.    **static** **const** unsigned **int** kBranching = 4;  
5.    **int** height = 1;  
6.    **while** (height < kMaxHeight && ((rnd\_.Next() % kBranching) == 0)) \{  
7.      height++;  
8.    \}  
9.    **return** height;  
10. \}

按照上面的代码，我们可以得出，建立x级索引的概率是0.25 ^(x - 1) \* 0.75，所以，建立1级索引的概率为75%，建立2级索引的概率为25%\*75%=18.75%，...（个人感觉，google把分支因子定为4有点高了，这样在绝大多数情况下，跳表的高度都不大于3）。

4.  在level0 ~ level (x-1)中链表的合适位置插入record3，假定根据上面的公式，我们得到需要为record3建立2级索引，即x=2，因此需要在level0与level1中的链中插入record3：在level 0的链中，record3插在record2与record4之间，在level 1的链中，record3插入在record2与record5之间，形成了现在的索引结构，在查询一个记录时，可以从最高一级索引向下查找，节约比较次数。

## 2.5 Record Format ##

Level DB将用户的每个更新或删除操作组合成一个Record，其格式如下：

![1335533865_8142.png][]

从图中可以看出，每个Record会在原用key的基础上添加版本号以及key的类型（更新 or 删除），组成internal key。插入跳表时，是按照internal key进行排序，而非用户key。这样，我们只可能向跳表中添加节点，而不可能删除和替换节点。

Internal Key在比较时，按照下面的[算法][Link 1]：

**\[cpp\]**  [view plain][]   [copy][view plain]

1.  **int** InternalKeyComparator::Compare(**const** Slice& akey, **const** Slice& bkey) **const** \{  
2.    **int** r = user\_comparator\_->Compare(ExtractUserKey(akey), ExtractUserKey(bkey));  
3.    **if** (r == 0) \{  
4.      // 比较后面8个字节构造的整数，第一个字节的type为Least Significant Byte  
5.      **const** uint64\_t anum = DecodeFixed64(akey.data() + akey.size() - 8);  
6.      **const** uint64\_t bnum = DecodeFixed64(bkey.data() + bkey.size() - 8);  
7.      **if** (anum > bnum)  // 注意：整数大反而key比较小  
8.      \{  
9.        r = -1;  
10.     \} **else** **if** (anum < bnum) \{  
11.       r = +1;  
12.     \}  
13.   \}  
14.   **return** r;  
15. \}

根据上面的算法，我们可以得知Internal Key的比较顺序：

1. 如果User Key不相等，则User Key比较小的记录的Internal Key也比较小，User Key默认采用字典序（lexicographic）进行比较，可以在建表参数中自定义comparator。  
2. 如果type也相同，则比较Sequence Num，Sequence Num大的Internal Key比较小。  
3. 如果Sequence Num相等，则比较Type，type为更新（Key Type=1）的记录比的type为删除（Key Type=0）的记录的Internal Key小。  
  
在插入到跳表时，一般不会出现Internal Key相等的情况（除非在一个Batch中操作了同一条记录两次，这里会出现一种bug：在一个Write Batch中，先插入一条记录，然后删除这条记录，最后把这个Batch写入DB，会发现DB中这条记录存在。因此，不推荐在Batch中多次操作相同key的记录），User Key相同的记录插入跳表时，Sequence Num大的记录会排在前面。  
设计Internal Key有个以下一些作用：  
1. Level DB支持快照查询，即查询时指定快照的版本号，查询出创建快照时某个User Key对应的Value，那么可以组成这样一个Internal Key：Sequence=快照版本号，Type=1，User Key为用户指定Key，然后查询数据文件与内存，找到大于等于此Internal Key且User Key匹配的第一条记录即可（即Sequence Num小于等于快照版本号的第一条记录）。  
2.如果查询最新的记录时，将Sequence Num设置为0xFFFFFFFFFFFFFF即可。因为我们更多的是查询最新记录，所以让Sequence Num大的记录排前面，可以在遍历时遇见第一条匹配的记录立即返回，减少往后遍历的次数。

# 3.文件结构 #

## 3.1 文件组成 ##

Level DB包含一下几种文件：

<table> 
 <tbody> 
  <tr> 
   <td>文件类型</td> 
   <td>说明</td> 
  </tr> 
  <tr> 
   <td>dbname/MANIFEST-[0-9]+ &nbsp;</td> 
   <td>清单文件 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;</td> 
  </tr> 
  <tr> 
   <td>dbname/[0-9]+.log</td> 
   <td>db日志文件</td> 
  </tr> 
  <tr> 
   <td>dbname/[0-9]+.sst</td> 
   <td>dbtable文件</td> 
  </tr> 
  <tr> 
   <td>dbname/[0-9]+.dbtmp</td> 
   <td>db临时文件</td> 
  </tr> 
  <tr> 
   <td>dbname/CURRENT&nbsp;</td> 
   <td>记录当前使用的清单文件名</td> 
  </tr> 
  <tr> 
   <td>dbname/LOCK &nbsp;</td> 
   <td>DB锁文件</td> 
  </tr> 
  <tr> 
   <td>dbname/LOG</td> 
   <td>info log日志文件</td> 
  </tr> 
  <tr> 
   <td>dbname/LOG.old</td> 
   <td>旧的info log日志文件</td> 
  </tr> 
 </tbody> 
</table>

上面的log文件，sst文件，临时文件，清单文件末尾都带着序列号，序号是单调递增的（随着next\_file\_number从1开始递增），以保证不会和之前的文件名重复。另外，注意区分db log与info log：前者是为了防止保障数据安全而实现的二进制Log，后者是打印引擎中间运行状态及警告等信息的文本log。

随着更新与Compaction的进行，Level DB会不断生成新文件，有时还会删除老文件，所以需要一个文件来记录文件列表，这个列表就是清单文件的作用，清单会不断变化，DB需要知道最新的清单文件，必须将清单准备好后原子切换，这就是CURRENT文件的作用，Level DB的清单过程更新如下：

1. 递增清单序号，生成一个新的清单文件。

2. 将此清单文件的名称写入到一个临时文件中。  
3. 将临时文件rename为CURRENT。

代码如下：

**\[cpp\]**  [view plain][]   [copy][view plain]

1.  Status SetCurrentFile(Env\* env, **const** std::string& dbname,  
2.                        uint64\_t descriptor\_number) \{  
3.    // 创建一个新的清单文件名  
4.    std::string manifest = DescriptorFileName(dbname, descriptor\_number);  
5.    Slice contents = manifest;  
6.    // 移除"dbname/"前缀  
7.    assert(contents.starts\_with(dbname + "/"));  
8.    contents.remove\_prefix(dbname.size() + 1);  
9.    // 创建一个临时文件  
10.   std::string tmp = TempFileName(dbname, descriptor\_number);  
11.   // 写入清单文件名  
12.   Status s = WriteStringToFile(env, contents.ToString() + "\\n", tmp);  
13.   **if** (s.ok()) \{  
14.     // 将临时文件改名为CURRENT  
15.     s = env->RenameFile(tmp, CurrentFileName(dbname));  
16.   \}  
17.   **if** (!s.ok()) \{  
18.     env->DeleteFile(tmp);  
19.   \}  
20.   **return** s;  
21. \}

## 3.2 Manifest    ##

在介绍其他文件格式前，先了解清单文件，MANIFEST文件是Level DB的元信息文件，它主要包括下面一些信息：  
1. Comparator的名称

其的格式如下：  
![1335865966_8376.png][]

我们可以看看其序列化的代码：

**\[cpp\]**  [view plain][]   [copy][view plain]

1.  **void** VersionEdit::EncodeTo(std::string\* dst) **const** \{  
2.    **if** (has\_comparator\_) \{  // 记录Comparator名称  
3.      PutVarint32(dst, kComdparator);  
4.      PutLengthPrefixedSlice(dst, comparator\_);  
5.    \}  
6.    **if** (has\_log\_number\_) \{  // 记录Log Numer  
7.      PutVarint32(dst, kLogNumber);  
8.      PutVarint64(dst, log\_number\_);  
9.    \}  
10.   **if** (has\_prev\_log\_number\_) \{  // 记录Prev Log Number，现在已废弃，一般为0  
11.     PutVarint32(dst, kPrevLogNumber);  
12.     PutVarint64(dst, prev\_log\_number\_);  
13.   \}  
14.   **if** (has\_next\_file\_number\_) \{  // 记录下一个文件序号  
15.     PutVarint32(dst, kNextFileNumber);  
16.     PutVarint64(dst, next\_file\_number\_);  
17.   \}  
18.   **if** (has\_last\_sequence\_) \{  // 记录最大的sequence num  
19.     PutVarint32(dst, kLastSequence);  
20.     PutVarint64(dst, last\_sequence\_);  
21.   \}  
22.   // 记录每一级Level下次compaction的起始Key  
23.   **for** (**size\_t** i = 0; i < compact\_pointers\_.size(); i++) \{  
24.     PutVarint32(dst, kCompactPointer);  
25.     PutVarint32(dst, compact\_pointers\_\[i\].first);  // level  
26.     PutLengthPrefixedSlice(dst, compact\_pointers\_\[i\].second.Encode());  
27.   \}  
28.   // 记录每一级需要删除的文件  
29.   **for** (DeletedFileSet::const\_iterator iter = deleted\_files\_.begin();  
30.        iter != deleted\_files\_.end();  
31.        ++iter) \{  
32.     PutVarint32(dst, kDeletedFile);  
33.     PutVarint32(dst, iter->first);   // level  
34.     PutVarint64(dst, iter->second);  // file number  
35.   \}  
36.   // 记录每一级需要有效的sst以及其smallest与largest的key  
37.   **for** (**size\_t** i = 0; i < new\_files\_.size(); i++) \{  
38.     **const** FileMetaData& f = new\_files\_\[i\].second;  
39.     PutVarint32(dst, kNewFile);  
40.     PutVarint32(dst, new\_files\_\[i\].first);  // level  
41.     PutVarint64(dst, f.number);  
42.     PutVarint64(dst, f.file\_size);  
43.     PutLengthPrefixedSlice(dst, f.smallest.Encode());  
44.     PutLengthPrefixedSlice(dst, f.largest.Encode());  
45.   \}  
46. \}

## 3.3 Sortedtable ##

Level DB间歇性地将内存中的SkipList对应的数据集合Dump到磁盘上，生成一个sst的文件，这个文件的格式如下：

![1335525886_5008.png][]

按照SSTable的结构，可以正向遍历，也可以逆向遍历，但是逆向遍历的代价要远远高于正向遍历的代价，因为每条record都是变长的，且其没有记录前一条记录的偏移，因此逆向Group遍历时，只能先回到group（代码中称为一个restart，为了便于理解，下面都称为group）开头（一个Data Block的group一般为16条记录，每个Data Block的尾部有group起始位置偏移索引），然后从头开始正向遍历，直至找到其前一条记录，如果当前位置为group的第一条记录，则需要回到上一个group的开头，遍历到其最后一条记录。另外，内存中跳表反向的遍历效率也远远不如正向遍历。

## 3.4 Sparse Index ##

一个sst文件内部除了Data Block，还有Index Block，Index Block的结构与Data Block一样，只不过每个group只包含一条记录，即Data Block的最大Key与偏移。其实这里说最大Key并不是很准确，理论上，只要保存最大Key就可以实现二分查找，但是Level DB在这里做了个优化，它并保存最大key，而是保存一个能分隔两个Data Block的最短Key，如：假定Data Block1的最后一个Key为“abcdefg”，Data Block2的第一个Key为“abzxcv”，则index可以记录Data Block1的索引key为“abd”；这样的分割串可以有很多，只要保证Data Block1中的所有Key都小于等于此索引，Data Block2中的所有Key都大于此索引即可。这种优化缩减了索引长度，查询时可以有效减小比较次数。我们可以看看默认comparator如何实现这种分割的：

**\[cpp\]**  [view plain][]   [copy][view plain]

1.  **void** BytewiseComparatorImpl::FindShortestSeparator(  
2.        std::string\* start,  
3.        **const** Slice& limit) **const** \{  
4.      // 先比较获得最大公共前缀  
5.      **size\_t** min\_length = std::min(start->size(), limit.size());  
6.      **size\_t** diff\_index = 0;  
7.      **while** ((diff\_index < min\_length) &&  
8.             ((\*start)\[diff\_index\] == limit\[diff\_index\])) \{  
9.        diff\_index++;  
10.     \}  
11.     **if** (diff\_index >= min\_length) \{  
12.       // 如果start就是limit的前缀，则只能使用start本身作为分割  
13.     \} **else** \{  
14.       uint8\_t diff\_byte = **static\_cast**<uint8\_t>((\*start)\[diff\_index\]);  
15.       // 将第一个不同字符+1，并确保其不会溢出，同时比limit小  
16.       **if** (diff\_byte < **static\_cast**<uint8\_t>(0xff) &&  
17.           diff\_byte + 1 < **static\_cast**<uint8\_t>(limit\[diff\_index\])) \{  
18.         (\*start)\[diff\_index\]++;  
19.         start->resize(diff\_index + 1);  
20.         assert(Compare(\*start, limit) < 0);  
21.       \}  
22.     \}  
23.   \}

从上面可以看出，FindShortestSeparator方法并不严格，有些时候没有找出最短分割的key（比如第一个不等的字符已经为0xFF时），它只是一种优化，我们自定义Comparator时，既可以实现，也可以不实现，如果不实现，将始终使用Data Block的最大Key作为索引，并不影响功能正确性。

# 4. Operations #

在介绍了数据结构后，我们看看Level DB一些基本操作的实现：

## 4.1 创建一个新表    ##

创建一个新的表大概分为几步，包括建立各类文件以及内存中的[数据结构][Link 1]，线程同步对象等，关键代码如下：

**\[cpp\]**  [view plain][]   [copy][view plain]

1.  // DBImpl在构造时会初始化互斥体与信号量，创建一个空的memtable，并根据配置设置Comparator及LRU缓冲  
2.  DBImpl::DBImpl(**const** Options& options, **const** std::string& dbname)  
3.      : env\_(options.env),  
4.        internal\_comparator\_(options.comparator), // 初始化Comparator  
5.        options\_(SanitizeOptions(dbname, &internal\_comparator\_, options)),  // 检查参数是否合法  
6.        owns\_info\_log\_(options\_.info\_log != options.info\_log),  // 是拥有自己info log，还是使用用户提供的  
7.        owns\_cache\_(options\_.block\_cache != options.block\_cache), // 是否拥有自己的LRU缓冲，或者使用用户提供的  
8.        dbname\_(dbname),  // 数据表名称  
9.        db\_lock\_(NULL),  // 不创建也不锁定文件锁  
10.       shutting\_down\_(NULL),   
11.       bg\_cv\_(&mutex\_),  // 用于与后台线程交互的条件信号  
12.       mem\_(**new** MemTable(internal\_comparator\_)), // 创建一个新的跳表  
13.       imm\_(NULL),  // 用于双缓冲的缓冲跳表开始时为NULL  
14.       logfile\_(NULL),  // log文件  
15.       logfile\_number\_(0), // log文件的序号  
16.       log\_(NULL),  // log writer  
17.       logger\_(NULL),  // 用于在多线程环境中记录Owner logger的一个指针  
18.       logger\_cv\_(&mutex\_), // 用于与Logger交互的条件信号  
19.       bg\_compaction\_scheduled\_(**false**), // 没打开表时不起动后台的compaction线程  
20.       manual\_compaction\_(NULL) \{  
21.   // 增加memtable的引用计数  
22.   mem\_->Ref();  
23.   has\_imm\_.Release\_Store(NULL);  
24.   
25.   // 根据Option创建一个LRU的缓冲对象，如果options中指定了Cache空间，则使用用户  
26.   // 提供的Cache空间，否则会在内部确实创建8MB的Cache，另外，LRU的Entry数目不能超过max\_open\_files-10  
27.   **const** **int** table\_cache\_size = options.max\_open\_files - 10;  
28.   table\_cache\_ = **new** TableCache(dbname\_, &options\_, table\_cache\_size);  
29.   
30.   // 创建一个Version管理器  
31.   versions\_ = **new** VersionSet(dbname\_, &options\_, table\_cache\_,  
32.                              &internal\_comparator\_);  
33. \}  
34.   
35. Options SanitizeOptions(**const** std::string& dbname,  
36.                         **const** InternalKeyComparator\* icmp,  
37.                         **const** Options& src) \{  
38.   Options result = src;  
39.   result.comparator = icmp;  
40.   ClipToRange(&result.max\_open\_files,           20,     50000);  
41.   ClipToRange(&result.write\_buffer\_size,        64<<10, 1<<30);  
42.   ClipToRange(&result.block\_size,               1<<10,  4<<20);  
43.   // 如果用户未指定info log文件（用于打印状态等文本信息的日志文件），则由引擎自己创建一个info log文件。  
44.   **if** (result.info\_log == NULL) \{  
45.     // Open a log file in the same directory as the db  
46.     src.env->CreateDir(dbname);  // 如果目录不存在则创建  
47.     // 如果已存在以前的info log文件，则将其改名为LOG.old，然后创建新的log文件与日志的writer  
48.     src.env->RenameFile(InfoLogFileName(dbname), OldInfoLogFileName(dbname));  
49.     Status s = src.env->NewLogger(InfoLogFileName(dbname), &result.info\_log);  
50.     **if** (!s.ok()) \{  
51.       result.info\_log = NULL;  
52.     \}  
53.   \}  
54.   // 如果用户没指定LRU缓冲，则创建8MB的LRU缓冲  
55.   **if** (result.block\_cache == NULL) \{  
56.     result.block\_cache = NewLRUCache(8 << 20);  
57.   \}  
58.   **return** result;  
59. \}  
60.   
61. Status DBImpl::NewDB() \{  
62.   // 创建version管理器  
63.   VersionEdit new\_db;  
64.   // 设置Comparator  
65.   new\_db.SetComparatorName(user\_comparator()->Name());  
66.   new\_db.SetLogNumber(0);  
67.   // 下一个序号从2开始，1留给清单文件  
68.   new\_db.SetNextFile(2);  
69.   new\_db.SetLastSequence(0);  
70.   // 创建一个清单文件，MANIFEST-1  
71.   **const** std::string manifest = DescriptorFileName(dbname\_, 1);  
72.   WritableFile\* file;  
73.   Status s = env\_->NewWritableFile(manifest, &file);  
74.   **if** (!s.ok()) \{  
75.     **return** s;  
76.   \}  
77.   \{  
78.     // 写入清单文件头  
79.     log::Writer log(file);  
80.     std::string record;  
81.     new\_db.EncodeTo(&record);  
82.     s = log.AddRecord(record);  
83.     **if** (s.ok()) \{  
84.       s = file->Close();  
85.     \}  
86.   \}  
87.   **delete** file;  
88.   **if** (s.ok()) \{  
89.     // 设置CURRENT文件，使其指向清单文件  
90.     s = SetCurrentFile(env\_, dbname\_, 1);  
91.   \} **else** \{  
92.     env\_->DeleteFile(manifest);  
93.   \}  
94.   **return** s;

## 4.2 打开一个已存在的表 ##

上面的步骤中，其实还遗漏了一个的重要流程，那就是DB的Open方法。Level DB无论是创建表，还是打开现有的表，都是使用Open方法。代码如下：

**\[cpp\]**  [view plain][]   [copy][view plain]

1.  Status DB::Open(**const** Options& options, **const** std::string& dbname,  
2.                  DB\*\* dbptr) \{  
3.    \*dbptr = NULL;  
4.    
5.    DBImpl\* impl = **new** DBImpl(options, dbname);  
6.    impl->mutex\_.Lock();  
7.    VersionEdit edit;  
8.    // 如果存在表数据，则Load表数据，并对日志进行恢复，否则，创建新表  
9.    Status s = impl->Recover(&edit);  
10.   **if** (s.ok()) \{  
11.     // 从VersionEdit获取一个新的文件序号，所以如果是新建数据表，则第一个LOG的序号为2（1已经被MANIFEST占用）  
12.     uint64\_t new\_log\_number = impl->versions\_->NewFileNumber();  
13.     // 记录日志文件号，创建新的log文件及Writer对象  
14.     WritableFile\* lfile;  
15.     s = options.env->NewWritableFile(LogFileName(dbname, new\_log\_number),  
16.                                      &lfile);  
17.     **if** (s.ok()) \{  
18.       edit.SetLogNumber(new\_log\_number);  
19.       impl->logfile\_ = lfile;  
20.       impl->logfile\_number\_ = new\_log\_number;  
21.       impl->log\_ = **new** log::Writer(lfile);  
22.       // 如果存在原来的log，则回放log  
23.       s = impl->versions\_->LogAndApply(&edit, &impl->mutex\_);  
24.     \}  
25.     **if** (s.ok()) \{  
26.       // 删除废弃的文件（如果存在）  
27.       impl->DeleteObsoleteFiles();  
28.       // 检查是否需要Compaction，如果需要，则让后台启动Compaction线程  
29.       impl->MaybeScheduleCompaction();  
30.     \}  
31.   \}  
32.   impl->mutex\_.Unlock();  
33.   **if** (s.ok()) \{  
34.     \*dbptr = impl;  
35.   \} **else** \{  
36.     **delete** impl;  
37.   \}  
38.   **return** s;  
39. \}

从上面可以看出，其实到底是新建表还是打开表都是取决与DBImpl::Recover()这个方法的行为，它的流程如下：

**\[cpp\]**  [view plain][]   [copy][view plain]

1.  Status DBImpl::Recover(VersionEdit\* edit) \{  
2.    mutex\_.AssertHeld();  
3.    
4.    // 创建DB目录，不关注错误  
5.    env\_->CreateDir(dbname\_);  
6.    // 在DB目录下打开或创建（如果不存在）LOCK文件并锁定它，防止其他进程打开此表  
7.    Status s = env\_->LockFile(LockFileName(dbname\_), &db\_lock\_);  
8.    **if** (!s.ok()) \{  
9.      **return** s;  
10.   \}  
11.     
12.   **if** (!env\_->FileExists(CurrentFileName(dbname\_))) \{  
13.     // 如果DB目录下不存在CURRENT文件且允许在表不存在时创建表，则新建一个表返回  
14.     **if** (options\_.create\_if\_missing) \{  
15.       s = NewDB();  
16.       **if** (!s.ok()) \{  
17.         **return** s;  
18.       \}  
19.     \} **else** \{  
20.       **return** Status::InvalidArgument(  
21.           dbname\_, "does not exist (create\_if\_missing is false)");  
22.     \}  
23.   \} **else** \{  
24.     **if** (options\_.error\_if\_exists) \{  
25.       **return** Status::InvalidArgument(  
26.           dbname\_, "exists (error\_if\_exists is true)");  
27.     \}  
28.   \}  
29.   // 如果运行到此，表明表已经存在，需要load，第一步是从MANIFEST文件中恢复VersionSet  
30.   s = versions\_->Recover();  
31.   **if** (s.ok()) \{  
32.     SequenceNumber max\_sequence(0);  
33.     // 获取MANIFEST中获取最后一次持久化清单时在使用LOG文件序号，注意：这个LOG当时正在使用，  
34.     // 表明数据还在memtable中，没有形成sst文件，所以数据恢复需要从这个LOG文件开始(包含这个LOG)。  
35.     // 另外，prev\_log是早前版本level\_db使用的机制，现在以及不再使用，这里只是为了兼容  
36.     **const** uint64\_t min\_log = versions\_->LogNumber();  
37.     **const** uint64\_t prev\_log = versions\_->PrevLogNumber();  
38.     // 扫描DB目录，记录下所有比MANIFEST中记录的LOG更加新的LOG文件  
39.     std::vector<std::string> filenames;  
40.     s = env\_->GetChildren(dbname\_, &filenames);  
41.     **if** (!s.ok()) \{  
42.       **return** s;  
43.     \}  
44.     uint64\_t number;  
45.     FileType type;  
46.     std::vector<uint64\_t> logs;  
47.     **for** (**size\_t** i = 0; i < filenames.size(); i++) \{  
48.       **if** (ParseFileName(filenames\[i\], &number, &type)  
49.           && type == kLogFile  
50.           && ((number >= min\_log) || (number == prev\_log))) \{  
51.         logs.push\_back(number);  
52.       \}  
53.     \}  
54.     // 将LOG文件安装从小到大排序  
55.     std::sort(logs.begin(), logs.end());  
56.     // 逐个LOG文件回放    for (size\_t i = 0; i < logs.size(); i++) \{   
57.       // 回放LOG时，记录被插入到memtable，如果超过write buffer，则还会dump出level 0的sst文件，  
58.       // 此方法会将日志种每条记录的sequence num与max\_sequence进行比较，以记录下最大的sequence num。  
59.       s = RecoverLogFile(logs\[i\], edit, &max\_sequence);  
60.       // 更新最大的文件序号，因为MANIFEST文件中没有记录这些LOG文件占用的序号；  
61.       // 当然，也可能LOG的序号小于MANIFEST中记录的最大文件序号，这时不需要更新。  
62.       versions\_->MarkFileNumberUsed(logs\[i\]);  
63.     \}  
64.     **if** (s.ok()) \{  
65.       // 比较日志回放前后的最大sequence num，如果回放记录中有超过LastSequence()的记录，则替换  
66.       **if** (versions\_->LastSequence() < max\_sequence) \{  
67.         versions\_->SetLastSequence(max\_sequence);  
68.       \}  
69.     \}  
70.   \}  
71.   **return** s;  
72. \}

## 4.3 关闭一个已打开的表 ##

Level DB设计成只要删除DB对象就可以关闭表，其关键流程如下：

**\[cpp\]**  [view plain][]   [copy][view plain]

1.  DBImpl::~DBImpl() \{  
2.    // 通知后台线程，DB即将关闭  
3.    mutex\_.Lock();  
4.    // 后台线程会间歇性地检查shutting\_down\_对象的指针，一旦不为NULL就会退出  
5.    shutting\_down\_.Release\_Store(**this**);  
6.    // 注意:这里必须循环通知，直至compaction线程获得信号并设置了bg\_compaction\_scheduled\_为false    
7.    **while** (bg\_compaction\_scheduled\_) \{  
8.      bg\_cv\_.Wait();  
9.    \}  
10.   mutex\_.Unlock();  
11.   
12.   // 如果锁定了文件锁，则释放文件锁  
13.   **if** (db\_lock\_ != NULL) \{  
14.     env\_->UnlockFile(db\_lock\_);  
15.   \}  
16.   
17.   **delete** versions\_;  
18.   // 减去memtable的引用计数  
19.   **if** (mem\_ != NULL) mem\_->Unref();  
20.   **if** (imm\_ != NULL) imm\_->Unref();  
21.   // 销毁db log相关对象以及表缓冲对象  
22.   **delete** log\_;  
23.   **delete** logfile\_;  
24.   **delete** table\_cache\_;  
25.   
26.   // 如果info log与cache是引擎自己构建，则需要销毁它们  
27.   **if** (owns\_info\_log\_) \{  
28.     **delete** options\_.info\_log;  
29.   \}  
30.   **if** (owns\_cache\_) \{  
31.     **delete** options\_.block\_cache;  
32.   \}  
33. \}

由上可见，delete一个db对象可能会阻塞调用线程一段时间，必须让其完成一些必须完成的工作，才能进一步保障数据的安全。

## 4.4 随机查询 ##

Level DB可能dump多个sst文件，这些文件的key范围可能重叠。按照Level DB的设计，其会将sst分为7个等级，可以视为代龄，其中，只有Level 0中的sst可能存在key的区间重叠的情况，而level1 - level6中，同一level中的sst可以保证不重叠，但不同level之间的sst依然可能key重叠。因此，如果查询一个key，其最多可能在6+n个sst中同时存在，n为level0中sst的个数；同时，由于这些文件的生成有先后关系，查询时还需要注意顺序，Get一个key的流程如下：

![1335772471_9154.png][]

文章来源：http://blog.csdn.net/icefireelf/article/details/7515816

[LevelDB]: http://blog.csdn.net/icefireelf/article/details/7515816
[http_code.google.com_p_leveldb]: http://code.google.com/p/leveldb/
[view plain]: http://blog.csdn.net/icefireelf/article/details/7515816#
[1335447663_7043.png]: https://img-my.csdn.net/uploads/201204/26/1335447663_7043.png
[1335447972_2102.png]: https://img-my.csdn.net/uploads/201204/26/1335447972_2102.png
[1335523511_6056.png]: https://img-my.csdn.net/uploads/201204/27/1335523511_6056.png
[1335533865_8142.png]: https://img-my.csdn.net/uploads/201204/27/1335533865_8142.png
[Link 1]: http://lib.csdn.net/base/31
[1335865966_8376.png]: https://img-my.csdn.net/uploads/201205/01/1335865966_8376.png
[1335525886_5008.png]: https://img-my.csdn.net/uploads/201204/27/1335525886_5008.png
[1335772471_9154.png]: https://img-my.csdn.net/uploads/201204/30/1335772471_9154.png