LevelDB顺序存储操作Golang模块封装及持久化队列实现

LevelDB介绍

Leveldb是一个google实现的非常高效的kv数据库，能够支持billion级别的数据量。 在这个数量级别下还有着非常高的性能。

LevelDB 是单进程的服务，性能非常之高，在一台4个Q6600的CPU机器上，每秒钟写数据超过40w，而随机读的性能每秒钟超过10w。

内部LSM 树算法实现。

LSM 大致结构如上图所示。

LSM 树而且通过批量存储技术规避磁盘随机写入问题。 LSM 树的设计思想非常朴素, 它的原理是把一颗大树拆分成N棵小树， 它首先写入到内存中（内存没有寻道速度的问题，随机写的性能得到大幅提升），在内存中构建一颗有序小树，随着小树越来越大，内存的小树会flush到磁盘上。磁盘中的树定期可以做 merge 操作，合并成一棵大树，以优化读性能【读数据的过程可能需要从内存 memtable 到磁盘 sstfile 读取多次，称之为读放大】。LevelDB 的 LSM 体现在多 level 文件格式上，最热最新的数据尽在 L0 层，数据在内存中，最冷最老的数据尽在 LN 层，数据在磁盘或者固态盘上。还有一种日志文件叫做 manifest，用于记录对 sstfile 的更改，可以认为是 LevelDB 的 GIF。

LevelDB是Google的 Jeff Dean和Sanjay Ghemawat设计开发的key-value存储引擎。

LevelDB底层存储利用了LSM tree的思想， RocksDB是Facebook基于LevelDB开发的存储引擎，针对LevelDB做了很多优化，但是大部分模块的实现机制是一样的。

LevelDB是一个持久化存储的KV系统，和Redis这种内存型的KV系统不同，LevelDB不会像Redis一样狂吃内存，而是将大部分数据存储到磁盘上。LevleDB在存储数据时，是根据记录的key值有序存储的，就是说相邻的key值在存储文件中是依次顺序存储的，而应用可以自定义key大小比较函数，LevleDB会按照用户定义的比较函数依序存储这些记录。

像大多数KV系统一样，LevelDB的操作接口简单，基本操作包括写记录、读记录以及删除记录。另外，LevelDB支持数据快照（snapshot）功能，使得读取操作不受写操作影响，可以在读操作过程中始终看到一致的数据。除此之外，LevelDB还支持数据压缩等操作，这对于减小存储空间以及增快IO效率都有直接的帮助。

对LevelDB进一步封装的理由

虽然LevelDB很强大，但是有点儿底层的味道了，操作不够友好，就几个简单的put,get接口。比如想实现个持久化的顺序操作的队列，想按顺序存储和读取记录，比如不用关心底层的操作，仅使用接口。比如如果哪天想替换底层存储，可以灵活一点儿，不用改动业务......

这里对LevelDB记录存储做一个封装，并实现了持久化的消息队列。同时也可以用在嵌入式linux上，替代sqllite做一些记录存储的模块功能。

封装后可供使用的接口：

// Recorder 操作记录的接口声明
```
type Recorder interface {
// 初始化记录区(会清空所有数据!)

InitRecAreas() error
// 打开记录区(开机必须先打开一次)
OpenRecAreas() (err error)
// 保存记录 areaID,存储区ID,取值从1--至--MAXRECAREAS(相当于一个表)
SaveRec(areaID RecArea, data interface{}, recType int) (id int, err error)
// 更新记录
UpdateRec(areaID RecArea, recID int, data interface{}, recType int) (id int, err error)
// 删除记录
DeleteRec(areaID RecArea, num int) (err error)
// 获取未上传记录数量
GetNoUploadNum(areaID RecArea) int
// 按数据库中的ID读取一条记录
ReadRecByID(areaID RecArea, id int) (p *Records, err error)
// 顺序读取未上传的记录
ReadRecNotServer(areaID RecArea, sn int) (p *Records, err error)
// 倒数读取记录（如sn=1代表最后一次写入的记录）
ReadRecWriteNot(areaID RecArea, sn int) (p *Records, err error)
// 最后一条记录流水
GetLastRecNO(areaID RecArea) int
// 获取当前的读、写ID
GetReadWriteID(areaID RecArea) (rid, wid int)
}
```

实现原理

单独维护一个key,类似于目录的概念，作为全局使用，记录当前的写的位置，读的位置，是否是循环覆盖写。

如图所示，RecDirTB相当于目录表，这里分配了三个对应可以有三个不同的表/存储区/队列供使用。

sn是记录流水号，wid当前写的位置，rid当前读（消费）的位置。flag是否循环覆盖写标识。

可指定容量，比如为100W,这样可做到记录循环覆盖写，记录留有底可查。

对应的记录区1的表名为Rec01TB,,后面的|xx为顺序存储的记录的id.

value为存储的json格式的内容。

格式为id,sn(记录流水),type（记录类型）,time（操作时间）,data（记录内容，任意的json格式数据），ext,res留作备注用。

测试的写入性能，封装后测试，并发写入1000条记录，总共耗时30ms左右。 

github地址：https://github.com/yangyongzhen/dbmod

简单的使用demo:

package main
import "fmt"
var (
recApi dbmod.Recorder
dataType int
)
// RequestQrcode模拟写入的数据内容json
type RequestQrcode struct {
ChanNO string `json:"chanNO"`
TermID string `json:"termID"`
Qrcode string `json:"qrcode"`
Money uint32 `json:"money"`
Recsn uint32 `json:"recsn"`
Orderno string `json:"orderno"`
Dealtime string `json:"dealtime"`
}
func main() {

//每次必须先打开存储区
err := recApi.OpenRecAreas()
if err != nil {
fmt.Printf("OpenRecAreas error,%s\n", err.Error())
}

data := RequestQrcode{}
data.ChanNO = "YS_CHANaaa"
data.TermID = "12345678"
data.Recsn = uint32(sn)
data.Qrcode = "6225882618789"
data.MOney= 1
dataType = 0x0A
// 按队列顺序写入一条记录,data为interface{},会序列化为json存储
id, err := recApi.SaveRec(dbmod.RecArea01, data, datatype)
if err != nil {
fmt.Printf("SaveRec error,%s\n", err.Error())
}
//按队列顺序读取一条记录
rec, err := recApi.ReadRecNotServer(dbmod.RecArea01, 1)
if err != nil {
fmt.Printf("ReadRecNotServer error,%s\n", err.Error())
}
fmt.Printf("rec:%#v\n", rec)

//按队列顺序删除一条记录(注:只更新标记)
recApi.DeleteRec(dbmod.RecArea01, 1)

//获取队列中未上传/消费的记录数量
num := recApi.GetNoUploadNum(dbmod.RecArea01)
fmt.Printf("GetNoUploadNum:%d\n", num)
}

 部分源码：

package dbmod
import (
"encoding/json"
"errors"
"fmt"
"log"
"sync"
"time"
)
var (
//IsDebug 是否调试
IsDebug = true
recDir [MAXRECAREAS]RecDir
lockSave = sync.Mutex{}
lockDel = sync.Mutex{}
once sync.Once
singleintance *Records
)
// Records ...
type Records struct {
ID int `json:"id"`
RecNo int `json:"sn"`
RecType int `json:"type" `
RecTime string `json:"time" `
Data interface{} `json:"data" `
Ext string `json:"ext" `
Res string `json:"res" `
}
// InitRecAreas 初始化记录存储区
func (rec Records) InitRecAreas() error {
//　清空数据
err := DelAllData()
if err != nil {
log.Fatal(err.Error())
return err
}
// 初始化目录表
err = InitRecDir()
if err != nil {
log.Fatal(err.Error())
return err
}
return err
}
// OpenRecAreas 打开记录存储区,每次开机，需要先打开一下
func (rec *Records) OpenRecAreas() (err error) {
//加载RecDir
for i := 0; i log.Printf("LoadDirs %02d \n", i+1)
err = recDir[i].LoadDirs(RecArea(i) + 1)
if err != nil {
log.Println(err.Error())
return
}
log.Printf("LoadDirs %02d ok!\n", i+1)
}
//log.Println(recDir)
return err
}
func saveRecToLDB(areaID RecArea, rec *Records, wid int) (id int, err error) {
t := time.Now()
rec.RecTime = t.Format("20060102150405")
rec.ID = wid
key := fmt.Sprintf("Rec%02dTB|%d", areaID, wid)
bv, err := json.Marshal(rec)
if err != nil {
log.Println("saveRecToLDB Marshal Error:", err)
return id, err
}
err = PutData(key, bv)
if err != nil {
log.Println("saveRecToLDB PutData Error:", err)
return id, err
}
return id, err
}
// SaveRec 保存记录
func (rec *Records) SaveRec(areaID RecArea, data interface{}, recType int) (id int, err error) {
lockSave.Lock()
defer lockSave.Unlock()
//log.Printf("SaveRec,area=%02d \n", areaID)
if (areaID <= 0) || (areaID > MAXRECAREAS) {
err = fmt.Errorf("area id %02d is not right,must between 1 and %02d", areaID, MAXRECAREAS)
log.Println(err.Error())
return
}
rec.RecNo = recDir[areaID-1].RecNo
rec.Data = data
rec.RecType = recType
//记录是否存储满，判断
if (recDir[areaID-1].WriteID + 1) > (MAXRECCOUNTS) {
if recDir[areaID-1].ReadID == 0 {
err = fmt.Errorf("rec area %02d is full", areaID)
log.Println(err.Error())
return
}
if (recDir[areaID-1].WriteID + 1 - (MAXRECCOUNTS)) == recDir[areaID-1].ReadID {
err = fmt.Errorf("rec area %02d is full", areaID)
log.Println(err.Error())
return
}
//保存记录
recDir[areaID-1].RecNo++
recDir[areaID-1].WriteID = 1
recDir[areaID-1].Flag = "1"
id = 1
_, err = saveRecToLDB(areaID, rec, id)
if err != nil {
log.Println("saveRecToLDB Error:", err)
return
}
err = recDir[areaID-1].UpdateDirs(areaID)
if err != nil {
log.Println("SaveRec UpdateDirs Error:", err)
return
}
//log.Printf("SaveRec,area=%02d ok!\n", areaID)
return id, err
}
if recDir[areaID-1].Flag == "1" {
//记录是否满判断
if (recDir[areaID-1].WriteID + 1) == recDir[areaID-1].ReadID {
err = fmt.Errorf("rec area %02d is full", areaID)
log.Println(err.Error())
return
}
rec.RecNo += 1
id = recDir[areaID-1].WriteID + 1
_, err = saveRecToLDB(areaID, rec, id)
if err != nil {
log.Println("saveRecToLDB Error:", err)
return
}
recDir[areaID-1].RecNo++
recDir[areaID-1].WriteID = id
err = recDir[areaID-1].UpdateDirs(areaID)
if err != nil {
log.Fatal(err.Error())
return 0, err
}
//log.Printf("SaveRec,area=%02d ok!\n", areaID)
return id, err
}
rec.RecNo += 1
id = recDir[areaID-1].WriteID + 1
_, err = saveRecToLDB(areaID, rec, id)
if err != nil {
log.Println("saveRecToLDB Error:", err)
return
}
recDir[areaID-1].RecNo++
recDir[areaID-1].WriteID = id
err = recDir[areaID-1].UpdateDirs(areaID)
if err != nil {
log.Fatal(err.Error())
return 0, err
}
//log.Printf("SaveRec,area=%02d ok!\n", areaID)
return id, err
}
// UpdateRec 更新记录
func (rec *Records) UpdateRec(areaID RecArea, recID int, data interface{}, recType int) (id int, err error) {
if (areaID <= 0) || (areaID > MAXRECAREAS) {
err = fmt.Errorf("area id %02d is not right,must between 1 and %02d", areaID, MAXRECAREAS)
log.Println(err.Error())
return
}
rec.Data = data
rec.RecType = recType
id, err = saveRecToLDB(areaID, rec, recID)
return id, err
}
// DeleteRec 删除记录（并不是真正删除表里记录，而是清除该记录的上传标记）
// areaID:记录区 num:删除的数量
func (rec Records) DeleteRec(areaID RecArea, num int) (err error) {
lockDel.Lock()
defer lockDel.Unlock()
if (areaID <= 0) || (areaID > MAXRECAREAS) {
err = errors.New("area id is not right")
log.Fatal(err.Error())
return
}
id := recDir[areaID-1].ReadID
//如果写的位置等于读的位置，说明记录已上传完，没有要删除的了
if recDir[areaID-1].WriteID == recDir[areaID-1].ReadID {
return
}
//如果要删除的数量大于了最大的记录数
if (id + num) > MAXRECCOUNTS {
if (id + num - MAXRECCOUNTS) > recDir[areaID-1].WriteID {
recDir[areaID-1].ReadID = recDir[areaID-1].WriteID
err = recDir[areaID-1].UpdateDirs(areaID)
if err != nil {
log.Fatal(err.Error())
return err
}
return
}
//更新读指针（读的位置）
recDir[areaID-1].ReadID = id + num - MAXRECCOUNTS
err = recDir[areaID-1].UpdateDirs(areaID)
if err != nil {
log.Fatal(err.Error())
return err
}
return
}
//如果当前写的位置大于读的位置
if recDir[areaID-1].WriteID > recDir[areaID-1].ReadID {
if id+num > recDir[areaID-1].WriteID {
//更新读指针（读的位置）
recDir[areaID-1].ReadID = recDir[areaID-1].WriteID
err = recDir[areaID-1].UpdateDirs(areaID)
if err != nil {
log.Fatal(err.Error())
return err
}
return
}
}
//更新读指针（读的位置）
recDir[areaID-1].ReadID = id + num
err = recDir[areaID-1].UpdateDirs(areaID)
if err != nil {
log.Fatal(err.Error())
return err
}
return
}
//GetNoUploadNum 获取未上传记录数量
func (rec Records) GetNoUploadNum(areaID RecArea) int {
num := 0
if recDir[areaID-1].WriteID == recDir[areaID-1].ReadID {
num = 0
return num
}
if recDir[areaID-1].Flag != "1" {
num = int(recDir[areaID-1].WriteID - recDir[areaID-1].ReadID)
} else {
if recDir[areaID-1].WriteID > recDir[areaID-1].ReadID {
num = int(recDir[areaID-1].WriteID - recDir[areaID-1].ReadID)
} else {
num = int(MAXRECCOUNTS - recDir[areaID-1].ReadID + recDir[areaID-1].WriteID)
}
}
return num
}
// ReadRecByID 按数据库ID读取记录
func (rec Records) ReadRecByID(areaID RecArea, rid int) (p *Records, err error) {
var rec1 Records
if (areaID <= 0) || (areaID > MAXRECAREAS) {
err = errors.New("area id is not right")
log.Fatal(err.Error())
return
}
key := fmt.Sprintf("Rec%02dTB|%d", areaID, rid)
bv, err := GetData(key)
err = json.Unmarshal(bv, &rec1)
if err != nil {
log.Println("ReadRecByID Unmarshal Error:", key, err)
}
return &rec1, nil
}
//ReadRecNotServer 读取未上传的记录数据，顺序读取第SN条未上传的记录
//sn取值 1-到-->未上传记录数目
func (rec Records) ReadRecNotServer(areaID RecArea, sn int) (p *Records, err error) {
if (areaID <= 0) || (areaID > MAXRECAREAS) {
err = errors.New("area id is not right")
log.Fatal(err.Error())
return
}
id := recDir[areaID-1].ReadID
//fmt.Printf("id=%d\n", id)
if (int(id) + sn) > MAXRECCOUNTS {
if int(id)+sn-MAXRECCOUNTS > int(recDir[areaID-1].WriteID) {
return nil, errors.New("no records")
}
p, err = rec.ReadRecByID(areaID, int(id)+sn-MAXRECCOUNTS)
} else {
if recDir[areaID-1].ReadID <= recDir[areaID-1].WriteID {
if (int(id) + sn) > int(recDir[areaID-1].WriteID) {
return nil, errors.New("no records")
}
}
p, err = rec.ReadRecByID(areaID, int(recDir[areaID-1].ReadID)+sn)
}
return p, err
}
// ReadRecWriteNot 倒数读取第SN条写入的记录
//读取一条记录 倒数读取第SN条写入的记录
func (rec Records) ReadRecWriteNot(areaID RecArea, sn int) (p *Records, err error) {
id := int(recDir[areaID-1].WriteID)
if (id - sn) <0 {
if recDir[areaID-1].Flag == "1" {
p, err = rec.ReadRecByID(areaID, MAXRECCOUNTS-(sn-id-1))
} else {
return nil, errors.New("no records")
}
} else {
p, err = rec.ReadRecByID(areaID, (id - sn + 1))
}
return
}
// GetLastRecNO 获取最后一条记录流水号
func (rec Records) GetLastRecNO(areaID RecArea) int {
if (areaID <= 0) || (areaID > MAXRECAREAS) {
log.Println("area id is not right")
return 0
}
id := recDir[areaID-1].RecNo
return id
}
// GetReadWriteID 获取当前的读、写ID
func (rec Records) GetReadWriteID(areaID RecArea) (rid, wid int) {
if (areaID <= 0) || (areaID > MAXRECAREAS) {
log.Println("area id is not right")
return 0, 0
}
rid = recDir[areaID-1].ReadID
wid = recDir[areaID-1].WriteID
return rid, wid
}
// NewRecords ...
func NewRecords(debug bool) *Records {
IsDebug = debug
if singleintance == nil {
once.Do(func() {
fmt.Println("Init singleintance Record operation ")
singleintance = new(Records)
})
}
return singleintance
}