Golang實現事務型內存數據庫的方法詳解

更新時間：2023年03月03日 11:44:29 作者：qingwave

內存數據庫經我們經常用到，例如Redis，那么如何從零實現一個內存數據庫呢，本文旨在介紹如何使用Golang編寫一個KV內存數據庫MossDB

特性

MossDB是一個純Golang編寫、可嵌入的、鍵值型內存數據庫，包含以下特性

可持久化，類似Redis AOF(Append only Log)
支持事務
支持近實時的TTL(Time to Live), 可以實現毫秒級的過期刪除
前綴搜索
Watch接口，可以監(jiān)聽某個鍵值的內容變化，類似etcd的Watch
多后端存儲，目前支持HashMap和RadixTree

命名由來

Moss有苔、苔花的含義，MossDB的名字來源于清代袁牧的一句詩:

苔花如米小，也學牡丹開

MossDB雖小，但五臟俱全，也支持了很多重要功能。另外，巧合的是《流浪地球2》中的超級計算機550W名字就是Moss。

架構

內存數據庫雖然使用簡單，實現起來卻有很多細節(jié)，Golang目前也存在不少優(yōu)秀的開源內存數據庫，比如buntdb、go-memdb，在編寫MossDB過程中也借鑒了一些它們的特性。

MossDB的架構如圖：

自上往下分為：

接口層，提供API接受用戶請求
核心層，實現事務、過期刪除、Watch等功能
存儲層，提供KV的后端存儲以及增刪改查
持久化層，使用AOL持久化即每次修改操作都會持久化到磁盤Log中

快速開始

MossDB可嵌入到Go程序中，可以通過go get獲取

go get github.com/qingwave/mossdb

MossDB提供了易用的API，可以方便地進行數據處理，下面的示例代碼展示了如何使用MossDB：

package main

import (
	"log"

	"github.com/qingwave/mossdb"
)

func main() {
    // create db instance
	db, err := mossdb.New(&mossdb.Config{})
	if err != nil {
		log.Fatal(err)
	}

    // set, get data
	db.Set("key1", []byte("val1"))
    log.Printf("get key1: %s", db.Get("key1"))

    // transaction
    db.Tx(func(tx *mossdb.Tx) error {
        val1 := tx.Get("key1")

        tx.Set("key2", val1)

        return nil
    })
}

更多示例見源碼

具體實現

從下往上分別介紹下MossDB如何設計與實現，以及相關的細節(jié)。

AOF持久化

AOF源于Redis提供兩種持久化技術，另外一種是RDB，AOF是指在每次寫操作后，將該操作序列化后追加到文件中，重啟時重放文件中的對應操作，從而達到持久化的目的。其實現簡單，用在MossDB是一個不錯的選擇，但需要注意的是AOF缺點同樣明顯，如果文件較大，每次重啟會花費較多時間。

Redis的AOF是一種后寫式日志，先寫內存直接返回給用戶，再寫磁盤文件持久化，可以保證其高性能，但如果中途宕機會丟失數據。MossDB中的AOF采用了WAL(預寫式日志)實現，先寫Log再寫內存，用來保證數據不會丟失，從而可以進一步實現事務。

那么采用WAL會不會影響其性能？每次必須等到落盤后才進行其他操作，WAL的每次寫入會先寫到內核緩沖區(qū)，這個調用很快就返回了，內核再將數據落盤。我們也可以使用fsync調用強制內核執(zhí)行直接將數據寫入磁盤。在MossDB中普通寫操作之會不會直接調用fsync，事務寫中強制開啟fsync，從而平衡數據一致性與性能。

WAL的實現引用了tiwall/wal，其中封裝了對Log文件的操作，可以支持批量寫入。由于WAL是二進制的，必須將數據進行編碼，通過varint編碼實現，將數據長度插入到數據本體之前，讀取時可以讀取定長的數據長度，然后按長度讀取數據本體。MossDB中數據格式如下：

type Record struct {
	Op        uint16
	KeySize   uint32
	ValSize   uint32
	Timestamp uint64
	TTL       uint64
	Key       []byte
	Val       []byte
}

對應編碼后的二進制格式為:

|------------------------------------------------------------|
| Op | KeySize | ValSize | Timestamp | TTL | Key | Val |
|------------------------------------------------------------|
| 2 | 4 | 4 | 8 | 8 | []byte | []byte |
|------------------------------------------------------------|

使用binary.BigEndian.PutUint16進行編碼，解碼時通過binary.BigEndian.Uint16，從而依次取得生成完整的數據。

存儲引擎

MossDB提供了存儲接口，只要實現了此接口都可以作為其后端存儲

type Store interface {
	Get(key string) ([]byte, bool)
	Set(key string, val []byte)
	Delete(key string)
	Prefix(key string) map[string][]byte
	Dump() map[string][]byte

	Len() int
}

內置提供了HashMap與RadixTree兩種方式，HashMap實現簡單通過簡單封裝map可以快速進行查詢與插入，但范圍搜索性能差。RadixTree即前綴樹，查詢插入的時間復雜度只與Key的長度相關，而且支持范圍搜索，MossDB采用Adaptive Radix Tree可以避免原生的前準樹空間浪費。

由于RadixTree的特性，MossDB可以方便的進行前綴搜索，目前支持List與Watch操作。

事務實現

要實現事務必須要保證其ACID特性，MossDB的事務定義如下：

type Tx struct {
	db      *DB // DB實例
	commits []*Record // 對于寫操作生成 Record, 用來做持久化
	undos   []*Record // 對于寫操作生成 Undo Record，用于回滾
}

MossDB中一次事務的流程主要包含以下幾個步驟：

首先加鎖，保證其數據一致性
對于寫操作，生成Commits和Undo Records，然后寫入內存；讀操作則直接執(zhí)行
提交階段，將Commits持久化到WAL中；若寫入失敗，則刪除已寫入數據；成功則設置數據的其他屬性(TTL, Watch等)
若中間發(fā)生錯誤，執(zhí)行回滾操作，將Undo Records的記錄執(zhí)行
事務完成，釋放鎖

Watch

由于工作中經常使用Kubernetes，對于其Watch接口印象深刻，通過Watch來充當其事件總線，保證其聲明式對象的管理。Kubernetes的Watch底層由etcd實現，MossDB也實現了類似的功能。

Watch的定義如下：

type Watcher struct {
	mu       sync.RWMutex // 鎖
	watchers map[string]*SubWatcher // watchId與具體Watcher直接的映射
	keys     map[string]containerx.Set[string] // Watch單個key
	ranges   *art.Tree // 前綴Watch
	queue    workqueue.WorkQueue // 工作隊列，存放所有事件
	stop     chan struct{} // 是否中止
}

通過工作隊列模式，任何寫操作都會同步追加到隊列中，如果存在單個key的監(jiān)聽者，則通過watchers map獲取到對應列表，依次發(fā)送事件。對于前綴Watch，我們不可能記錄此前綴的所有Key，這里借鑒了etcd，通過RadixTree保存前綴Key，當有新事件時，匹配Key所在的路徑，如果有監(jiān)聽者，則進行事件通知。

調用Watch會返回一個Channel，用戶只需要監(jiān)聽Channel即可

func Watch(db *mossdb.DB) {
	key := "watch-key"

	ctx, cancel := context.WithTimeout(context.Background(), 1000*time.Millisecond)
	defer cancel()

	// start watch key
	ch := db.Watch(ctx, key)
	log.Printf("start watch key %s", key)

	go func() { // 模擬發(fā)送event
		db.Set(key, mossdb.Val("val1"))
		db.Set(key, mossdb.Val("val2"))
		db.Delete(key)

		time.Sleep(100 * time.Second)

		db.Set(key, mossdb.Val("val3"))
	}()

	for {
		select {
		case <-ctx.Done():
			log.Printf("context done")
			return
		case resp, ok := <-ch:
			if !ok {
				log.Printf("watch done")
				return
			}
			log.Printf("receive event: %s, key: %s, new val: %s", resp.Event.Op, resp.Event.Key, resp.Event.Val)
		}
	}

    // Output
    // 2023/02/23 09:48:50 start watch key watch-key
	// 2023/02/23 09:34:19 receive event: MODIFY, key: watch-key, new val: val1
	// 2023/02/23 09:34:19 receive event: MODIFY, key: watch-key, new val: val2
	// 2023/02/23 09:34:19 receive event: DELETE, key: watch-key, new val:
	// 2023/02/23 09:34:19 context done
}