LevelDB 存储模块 Go 语言封装及高性能持久化队列实现

杨永贞

发布于 2021-04-12 17:28:08

1.4K0

发布于 2021-04-12 17:28:08

文章被收录于专栏：独行猫a的沉淀积累总结独行猫a的沉淀积累总结

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)
	// 保存记录，相当于入队push操作 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)
	// 删除记录(相当于出队pop操作)
	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 < MAXRECAREAS; 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
}

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2021-04-08 ，如有侵权请联系 cloudcommunity@tencent.com 删除

存储