1.什么是Memtable？Memtable是Rocksdb在内存中存储数据的一种数据结构。Memtable的容量是固定的。Memtable写满后会转为ImmutableMemtable，ImmutableMemtable中的数据会被flush到SSTFile中。Memtable和ImmutableMemtable的唯一区别是Memtable可以读写，而ImmutableMemtable是只读的，不允许写。RocksDB引入了ColumnFamily的概念。ColumnFamily中只有一个Memtable，但允许有多个ImmutableMemtable。Rocksdb支持创建多种数据结构类型的Memtable，默认是SkipList，也就是跳表。2、Memtable的数据结构Rocksdb中Memtable有多种实现（SkipList/HashSkipList/HashLinkList/Vector），默认实现是SkipList。一个Memtable中维护了两个SkipList，其中范围删除插入到range_del_table_，其他操作写入到table_。Memtable定义的操作接口Add()如下：boolMemTable::Add(SequenceNumbers,ValueTypetype,constSlice&key,/*userkey*/constSlice&value,boolallow_concurrent,MemTablePostProcessInfo*post_process_info){//a键值条目数据格式//key_size:varint32ofinternal_key.size()//keybytes:char[internal_key.size()]//value_size:varint32ofvalue.size()//valuebytes:char[value.size()]uint32_tkey_size=static_cast(key.size());uint32_tval_size=static_cast(value.size());uint32_tinternal_key_size=key_size+8;VarintLength(val_size)+val_size;char*buf=nullptr;//通过判断key-value类型选择memtable，将range_del_table_插入range_del_table_std::unique_ptr&table=type==kTypeRangeDeletion？range_del_table_:table_;KeyHandlehandle=table->Allocate(encoded_len,&buf);//...//是否允许并发插入if(!allow_concurrent){//是否制定函数提取key的前缀if(insert_with_hint_prefix_extractor_!=nullptr&&insert_with_hint_prefix_extractor_->InDomain(key_slice)){//...boolres=table->InsertWithHint(handle,&insert_hints_[prefix]);}else{//插入键值对boolres=table->Insert(handle);如果（不太可能（！res））{returnres;}}}else{//插入键值对boolres=table->InsertConcurrently(handle);如果（不太可能（！res））{returnres;}}returntrue;}Add()函数将用户的key和value封装到一个buf中，然后根据不同的条件调用table->Insert()插入到Memtable中。Table是Memtable的工厂类实现。默认是SkiplistRep，即key的插入是通过调用SkipList的Insert()来完成的。Memtable定义的操作接口Get()如下：boolMemTable::Get(constLookupKey&key,std::string*value,Status*s,MergeContext*merge_context,RangeDelAggregator*range_del_agg,SequenceNumber*seq,constReadOptions&read_opts,ReadCallback*callback,bool*is_blob_index){//在range_del_table_上初始化一个迭代器std::unique_ptrrange_del_iter(NewRangeTombstoneIterator(read_opts));状态status=range_del_agg->AddTombstones(std::move(range_del_iter));如果(!status.ok()){*s=状态;返回假；}切片user_key=key.user_key();//使用前缀提取和过滤判断key是否存在boolconstmay_contain=nullptr==prefix_bloom_Memtable调用Get()通过SkipList的FindGreaterOrEqual()最终找到SkipListRep的Get()接口。找到的key会传入回调函数SaveValu，e()会根据type进行处理，比如ktypeDeletion会返回NotFound()。3.什么是跳表？SkipList是跳表，在普通的单向链表上增加了一些索引，这些索引是有层次的，这样可以快速找到数据。下面是一个典型的跳表构建过程：最初我们有一个以首节点为首的有序链表a，然后为每两个相邻节点添加一个指针，让指针指向下一个节点，得到链表b。这样，所有新加入的指针都连接成了一个新的链表，但它包含的节点数只有原来的一半。然后我们对二级链表再做一遍得到c表。重复这一过程，直到只剩下一个采样节点，如图d所示。这样就完成了跳表的构建。跳表查找过程如下：从head开始，head的层级为4，判断head的后继节点的值小于<12，此时当前节点变为6，查找继续；节点6的层级为3，后继节点的值被判断为NIL，因此层级降为2；判断x->forward[2]->key(25)>17，继续降级为1；判断x->forward[1]->key(9)<17，此时x变为Forx->forward[1]，x变为节点9；节点9x->forward[1]->key的判断为17，所以找到节点直接返回。跳表插入过程如下：以上图为例，list->level=4，如果要插入节点17，首先确定查找路径，与前面的步骤类似。创建一个新节点Node(17)并为其生成一个关卡（随机）。级别的可能值为[1,MaxLevel]。这个时候就需要比较了。如果levellevel，需要把header突出的部分指向它，这里新生成的节点Node(17)的level为5，超过了当前list的最大level，所以update[4]设置为表头，然后直接将节点（17）作为表头的后继者。最后设置搜索路径上各个节点的后继关系，这样我们就完成了节点的插入。下面看一下SkipList的具体代码实现：InlineSkipList数据结构>>classInlineSkipList{private:structNode;结构拼接；public:usingDecodedKey=\typenamestd::remove_reference::type::DecodedType;…Allocator*constallocator_;比较器常量compare_；节点*常量头_；std::atomic最大高度_；拼接*seq_splice_;};Node的数据结构>>templatestructInlineSkipList::Node{private://存储节点的next_node数组//数组的大小为节点的高度。当调用NewNode()分配内存并初始化整个数组时std::atomicnext_[1];};Node的数据结构如下图，它存储了key和每一层的指针链表连续，通过next_[-n]访问每一层的next指针。另外，在创建新节点时，节点的高度会被写入next_[0]之前的4个字节处，当插入完成后，next_[0]会被恢复为指向同层下一个节点的指针.4、插入Memtable的InlineSkipList的Add()通过SkipList的Insert()查找。下面是Insert()的具体实现：boolInlineSkipList::Insert(constchar*key,Splice*splice,boolallow_partial_splice_fix){Node*x=reinterpret_cast(const_cast(key))-1;//x是next_[0]constDecodedKeykey_decoded=compare_.decode_key(key);intheight=x->UnstashHeight();断言（高度>=1&&高度<=kMaxHeight_）；intmax_height=max_height_.load(std::memory_order_relaxed);//更新max_heightwhile(height>max_height){if(max_height_.compare_exchange_weak(max_height,height)){//更新成功max_height=height;休息;}//否则，重试，可能会退出循环，因为是别人添加的}assert(max_height<=kMaxPossibleHeight);//插入节点时需要一个Splice对象，这个对象主要保存最新插入的节点快照//保存了一个prev和next节点指针数组，可以通过Level索引到Level对应的节点intrecompute_height=0;如果（拼接->高度_prev_[max_height]=head_;拼接->next_[max_height]=nullptr;拼接->height_=max_height;重新计算高度=最大高度；}else{while(recompute_heightprev_[recompute_height]->Next(recompute_height)!=splice->next_[recompute_height]){//判断这一层拼接是否紧密，即,prev_->Next是否等于next_++recompute_height；}elseif(splice->prev_[recompute_height]!=head_&&!KeyIsAfterNode(key_decoded,splice->prev_[recompute_height])){//小于当前层的prev_splice//...}elseif(KeyIsAfterNode(key_decoded,splice->next_[recompute_height])){//大于当前层splice的prev_//...}else{//找到了合适的levelbreak;}}}断言（重新计算高度<=最大高度）；if(recompute_height>0){//计算拼接RecomputeSpliceLevels(key_decoded,splice,recompute_height);//为要插入的密钥找到合适的拼接}boolsplice_is_valid=true;if(UseCAS){//CAS无锁机制//...}else{for(inti=0;i=recompute_height&&splice->prev_[i]->Next(i)!=splice->next_[i]){//确保splice的层级有效，如果无效则重新查找FindSpliceForLevel(key_decoded,splice->prev_[i],nullptr,i,&splice->prev_[i],&splice->next_[i]);}//如果(UNLIKELY(i==0&&splice->next_[i]!=nullptr&&compare_(x->Key(),splice->next_[i]->Key())>=0)){//重复键returnfalse;}if(不太可能(i==0&&splice->prev_[i]!=head_&&compare_(splice->prev_[i]->Key(),x->Key())>=0)){//重复键returnfalse;}//...x->NoBarrier_SetNext(i,splice->next_[i]);//将新节点指向对应的下一个节点splice->prev_[i]->SetNext(i,x);//将splice的prev节点的next指向新节点}}if(splice_is_valid){//将新节点高度以下的prev节点设置为该节点，因为原来的prev和next不连续向上for(inti=0;iprev_[i]=x;}//...}else{拼接->height_=0;}returntrue;}五、InlineSkipList查找Memtable的Get()通过SkipList的FindGreaterOrEqual()来查找，下面是FindGreaterOrEqual()的具体实现：InlineSkipList::FindGreaterOrEqual(constchar*key)const{Node*x=头_；intlevel=GetMaxHeight()-1;//从最高层开始查找Node*last_bigger=nullptr;constDecodedKeykey_decoded=compare_.decode_key(key);while(true){Node*next=x->Next(level);if(next!=nullptr){PREFETCH(next->Next(level),0,1);}//确保列表已排序assert(x==head_||next==nullptr||KeyIsAfterNode(next->Key(),x));//确保我们在搜索期间没有超出范围assert(x==head_||KeyIsAfterNode(key_decoded,x));intcmp=(next==nullptr||next==last_bigger)?1:compare_(next->Key(),key_decoded);if(cmp==0||(cmp>0&&level==0)){//找到相同的key或者搜索到的key不在这个范围内returnnext;}elseif(cmp<0){//待查找的key大于next，则在本层继续查找x=next;}else{//如果要查找的key不大于next，还没有到末尾，继续查找下一层//切换到nextlist，重复使用compare_()resultlast_bigger=next;等级-;}}}
