1.Batch的日常使用

batch的日常使用比较简单，使用Batch是提高leveldb写入性能的一个关键。

    db, _ := leveldb.OpenFile("file", nil)
    batch := leveldb.Batch{}
    batch.Put([]byte("k1"), []byte("y"))
    batch.Delete([]byte("k1"))
    db.Write(&batch, nil)

本文将结合batch的内部结构分析Put、Delete等操作的相关原理。

2.Batch的内部结构

db实例内部拥有一个batch的对象池，batchPool, 在常规的db.put操作中，实际数据也是插入到batch 中的：

    batch := db.batchPool.Get().(*Batch)
    batch.Reset()
    batch.appendRec(kt, key, value)
    return db.writeLocked(batch, batch, merge, sync)

使用batchPool的方式减少了batch对象的开销，batch的结构定义如下

type Batch struct {
    data  []byte
    index []batchIndex
    // internalLen is sums of key/value pair length plus 8-bytes internal key.
    internalLen int
}

batch使用一个byte数组存储具体数据，这里data存储实际的key和value值。batchIndex用于data内部的key,value的索引
Batch内部数据结构图示如下：

图1. batch internal datastructure.png

type batchIndex struct {
    keyType            keyType //key的类型 插入或者删除
    keyPos, keyLen     int //key的起始位置及其长度
    valuePos, valueLen int //value的起始长度以及其长度
}

3.Batch的关键函数

3.1 appendRec 数据插入函数

该内部函数是Batch的Put以及Delete函数的内部实现，Put或者Delete以key的类型进行区分。
batch的data部分存储key和value的长度，主要是为了decodeBatch的方便。

func (b *Batch) appendRec(kt keyType, key, value []byte) {
    //1.计算存储kt keylen  key valuelen value的data所需要的长度
    n := 1 + binary.MaxVarintLen32 + len(key)
    if kt == keyTypeVal {
        n += binary.MaxVarintLen32 + len(value)
    }

    //容量不足时及时扩充容量
    b.grow(n)

    index := batchIndex{keyType: kt}
    o := len(b.data)
    data := b.data[:o+n] //更新slice, 即增加了data的实际length
    data[o] = byte(kt)   // o位置加上标识位 占8位
    o++
    //序列化key值的长度并存储到data内，key最长不超过32位
    o += binary.PutUvarint(data[o:], uint64(len(key)))
    index.keyPos = o
    index.keyLen = len(key) //记录key的位置以及长度值

    o += copy(data[o:], key) //写入key
    if kt == keyTypeVal {
        o += binary.PutUvarint(data[o:], uint64(len(value)))
        index.valuePos = o
        index.valueLen = len(value)
        o += copy(data[o:], value) //写入value 返回copy的长度
    }
    b.data = data[:o]
    b.index = append(b.index, index)                   //添加一个batch索引项
    b.internalLen += index.keyLen + index.valueLen + 8 //内部batch的长度 key value长度再加上8位的标识位
}

3.2 grow 扩容函数

func (b *Batch) grow(n int) {
    o := len(b.data)
    if cap(b.data)-o < n {
        div := 1
        if len(b.index) > batchGrowRec { //3000
            div = len(b.index) / batchGrowRec
        }
        ndata := make([]byte, o, o+n+o/div)
              //当batch index很多的时候，扩容的扩充容量越小
              //batch index < batchGrowRec时扩充为 2*old cap + n的新长度
        copy(ndata, b.data)
        b.data = ndata
    }
}