golang常用库之-KV数据库之pebble

文章目录

golang常用库之-KV数据库之pebble
- pebble
- - 官方示例代码
- 实战
- pebble常用方法
- - NewSnapshot方法
  - NewIndexedBatch 方法
- 参考

golang常用库之-KV数据库之pebble

rocksdb是一款由Facebook使用C/C++开发的嵌入式的持久化的KV数据库。

Pebble 是 Cockroach 参考 RocksDB 并用 Go 语言开发的高性能 KV 存储引擎。

pebble

github地址：https://github.com/cockroachdb/pebble

Pebble 是一个受 LevelDB/RocksDB 启发的键值存储，专注于 CockroachDB 的性能和内部使用。Pebble 继承了 RocksDB 文件格式和一些扩展名，例如范围删除逻辑删除、表级绽放过滤器和 MANIFEST 格式更新。

Pebble 有意不追求在 RocksDB 中包含所有功能，而是专门针对 CockroachDB 所需的用例和功能集， RocksDB 有大量 Pebble 中没有实现的功能。

Pebble 对 RocksDB 进行了多项改进!

Pebble 在 CockroachDB v20.1（2020 年 5 月发布）中作为 RocksDB 的替代存储引擎引入，并在当时成功用于生产。Pebble 在 CockroachDB v20.2（2020 年 11 月发布）中成为默认存储引擎。Pebble 正在被 CockroachDB 的用户大规模地用于生产，并且被认为是稳定且可用于生产的。

官方示例代码

package mainimport ("fmt""log""github.com/cockroachdb/pebble"
)func main() {db, err := pebble.Open("demo", &pebble.Options{})if err != nil {log.Fatal(err)}key := []byte("hello")if err := db.Set(key, []byte("world"), pebble.Sync); err != nil {log.Fatal(err)}value, closer, err := db.Get(key)if err != nil {log.Fatal(err)}fmt.Printf("%s %s\n", key, value)if err := closer.Close(); err != nil {log.Fatal(err)}if err := db.Close(); err != nil {log.Fatal(err)}
}

实战

来自项目https://github.com/NethermindEth/juno 的
db/pebble/db.go

package pebbleimport ("sync""testing""github.com/NethermindEth/juno/db""github.com/NethermindEth/juno/utils""github.com/cockroachdb/pebble""github.com/cockroachdb/pebble/vfs"
)const (// minCache is the minimum amount of memory in megabytes to allocate to pebble read and write caching.minCache = 8
)var _ db.DB = (*DB)(nil)type DB struct {pebble   *pebble.DBwMutex   *sync.Mutexlistener db.EventListener
}// New opens a new database at the given path
func New(path string, cache uint, maxOpenFiles int, logger pebble.Logger) (db.DB, error) {// Ensure that the specified cache size meets a minimum threshold.cache = max(minCache, cache)pDB, err := newPebble(path, &pebble.Options{Logger:       logger,Cache:        pebble.NewCache(int64(cache * utils.Megabyte)),MaxOpenFiles: maxOpenFiles,})if err != nil {return nil, err}return pDB, nil
}// NewMem opens a new in-memory database
func NewMem() (db.DB, error) {return newPebble("", &pebble.Options{FS: vfs.NewMem(),})
}// NewMemTest opens a new in-memory database, panics on error
func NewMemTest(t *testing.T) db.DB {memDB, err := NewMem()if err != nil {t.Fatalf("create in-memory db: %v", err)}t.Cleanup(func() {if err := memDB.Close(); err != nil {t.Errorf("close in-memory db: %v", err)}})return memDB
}func newPebble(path string, options *pebble.Options) (*DB, error) {pDB, err := pebble.Open(path, options)if err != nil {return nil, err}return &DB{pebble: pDB, wMutex: new(sync.Mutex), listener: &db.SelectiveListener{}}, nil
}// WithListener registers an EventListener
func (d *DB) WithListener(listener db.EventListener) db.DB {d.listener = listenerreturn d
}// NewTransaction : see db.DB.NewTransaction
func (d *DB) NewTransaction(update bool) (db.Transaction, error) {txn := &Transaction{listener: d.listener,}if update {d.wMutex.Lock()txn.lock = d.wMutextxn.batch = d.pebble.NewIndexedBatch()} else {txn.snapshot = d.pebble.NewSnapshot()}return txn, nil
}// Close : see io.Closer.Close
func (d *DB) Close() error {return d.pebble.Close()
}// View : see db.DB.View
func (d *DB) View(fn func(txn db.Transaction) error) error {return db.View(d, fn)
}// Update : see db.DB.Update
func (d *DB) Update(fn func(txn db.Transaction) error) error {return db.Update(d, fn)
}// Impl : see db.DB.Impl
func (d *DB) Impl() any {return d.pebble
}

DB 结构体:
- pebble: Pebble 数据库的实例。
- wMutex: 用于控制写操作的互斥锁。
- listener: 数据库事件监听器。
New 函数:
- 用于打开一个给定路径的数据库。
- 确保指定的缓存大小满足最小阈值要求。
- 创建一个新的 Pebble 实例，并返回数据库实例。
newPebble 函数: 用于创建一个新的 Pebble 实例

func newPebble(path string, options *pebble.Options) (*DB, error) {pDB, err := pebble.Open(path, options)if err != nil {return nil, err}return &DB{pebble: pDB, wMutex: new(sync.Mutex), listener: &db.SelectiveListener{}}, nil
}

使用 pebble.Open 函数打开一个 Pebble 数据库，传入给定的路径和选项。
如果打开成功，则创建一个新的 DB 结构体实例，并初始化其中的字段：
pebble 字段使用刚刚打开的 Pebble 数据库实例。
wMutex 字段为一个新的互斥锁。
listener 字段为一个新的 db.SelectiveListener 实例的指针。

WithListener 方法:
用于注册一个事件监听器。
Close: 关闭数据库。
View: 执行数据库的只读操作。
Update: 执行数据库的读写操作。
Impl: 返回底层的 Pebble 实例。
NewTransaction: 创建一个新的事务。
NewTransaction 比较特殊是自定义业务有关的方法， “NewTransaction 返回一个在该数据库上的事务。如果请求创建一个更新事务,而另一个更新事务正在进行中,那么该方法应该会被阻塞。”

func (d *DB) NewTransaction(update bool) (db.Transaction, error) {txn := &Transaction{listener: d.listener,}if update {d.wMutex.Lock()txn.lock = d.wMutextxn.batch = d.pebble.NewIndexedBatch()} else {txn.snapshot = d.pebble.NewSnapshot()}return txn, nil
}

如果 update 为 true，则表示创建一个更新事务。此时，会获取 DB 对象的写入互斥锁 wMutex，将该互斥锁赋给事务对象的 lock 字段，并使用 Pebble 数据库的 NewIndexedBatch 方法创建一个新的批处理对象，并将其赋给事务对象的 batch 字段。
如果 update 为 false，则表示创建一个只读事务。此时，会使用 Pebble 数据库的 NewSnapshot 方法创建一个新的快照对象，并将其赋给事务对象的 snapshot 字段。

这样做的目的是为了保证数据库的一致性和隔离性。如果多个更新事务同时进行,可能会导致数据库状态出现不一致的情况。通过阻塞方式,可以确保同一时间只有一个更新事务在进行。

db/pebble/transaction.go

package pebbleimport ("errors""io""sync""time""github.com/NethermindEth/juno/db""github.com/NethermindEth/juno/utils""github.com/cockroachdb/pebble"
)var ErrDiscardedTransaction = errors.New("discarded txn")var _ db.Transaction = (*Transaction)(nil)type Transaction struct {batch    *pebble.Batchsnapshot *pebble.Snapshotlock     *sync.Mutexlistener db.EventListener
}// Discard : see db.Transaction.Discard
func (t *Transaction) Discard() error {if t.batch != nil {if err := t.batch.Close(); err != nil {return err}t.batch = nil}if t.snapshot != nil {if err := t.snapshot.Close(); err != nil {return err}t.snapshot = nil}if t.lock != nil {t.lock.Unlock()t.lock = nil}return nil
}// Commit : see db.Transaction.Commit
func (t *Transaction) Commit() error {start := time.Now()defer func() { t.listener.OnCommit(time.Since(start)) }()if t.batch != nil {return utils.RunAndWrapOnError(t.Discard, t.batch.Commit(pebble.Sync))}return utils.RunAndWrapOnError(t.Discard, ErrDiscardedTransaction)
}// Set : see db.Transaction.Set
func (t *Transaction) Set(key, val []byte) error {start := time.Now()if t.batch == nil {return errors.New("read only transaction")}if len(key) == 0 {return errors.New("empty key")}defer func() { t.listener.OnIO(true, time.Since(start)) }()return t.batch.Set(key, val, pebble.Sync)
}// Delete : see db.Transaction.Delete
func (t *Transaction) Delete(key []byte) error {start := time.Now()if t.batch == nil {return errors.New("read only transaction")}defer func() { t.listener.OnIO(true, time.Since(start)) }()return t.batch.Delete(key, pebble.Sync)
}// Get : see db.Transaction.Get
func (t *Transaction) Get(key []byte, cb func([]byte) error) error {start := time.Now()var val []bytevar closer io.Closervar err errorif t.batch != nil {val, closer, err = t.batch.Get(key)} else if t.snapshot != nil {val, closer, err = t.snapshot.Get(key)} else {return ErrDiscardedTransaction}defer t.listener.OnIO(false, time.Since(start))if err != nil {if errors.Is(err, pebble.ErrNotFound) {return db.ErrKeyNotFound}return err}return utils.RunAndWrapOnError(closer.Close, cb(val))
}// Impl : see db.Transaction.Impl
func (t *Transaction) Impl() any {if t.batch != nil {return t.batch}if t.snapshot != nil {return t.snapshot}return nil
}// NewIterator : see db.Transaction.NewIterator
func (t *Transaction) NewIterator() (db.Iterator, error) {var (iter *pebble.Iteratorerr  error)if t.batch != nil {iter, err = t.batch.NewIter(nil)if err != nil {return nil, err}} else if t.snapshot != nil {iter, err = t.snapshot.NewIter(nil)if err != nil {return nil, err}} else {return nil, ErrDiscardedTransaction}return &iterator{iter: iter}, nil
}