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

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 的用户大规模地用于生产，而且被认为是稳定且可用于生产的。
官方示例代码

1. package main
3. import (
4.         "fmt"
5.         "log"
7.         "github.com/cockroachdb/pebble"
8. )
10. func main() {
11.         db, err := pebble.Open("demo", &pebble.Options{})
12.         if err != nil {
13.                 log.Fatal(err)
14.         }
15.         key := []byte("hello")
16.         if err := db.Set(key, []byte("world"), pebble.Sync); err != nil {
17.                 log.Fatal(err)
18.         }
19.         value, closer, err := db.Get(key)
20.         if err != nil {
21.                 log.Fatal(err)
22.         }
23.         fmt.Printf("%s %s\n", key, value)
24.         if err := closer.Close(); err != nil {
25.                 log.Fatal(err)
26.         }
27.         if err := db.Close(); err != nil {
28.                 log.Fatal(err)
29.         }
30. }

复制代码

实战

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

1. package pebble
3. import (
4.         "sync"
5.         "testing"
7.         "github.com/NethermindEth/juno/db"
8.         "github.com/NethermindEth/juno/utils"
9.         "github.com/cockroachdb/pebble"
10.         "github.com/cockroachdb/pebble/vfs"
11. )
13. const (
14.         // minCache is the minimum amount of memory in megabytes to allocate to pebble read and write caching.
15.         minCache = 8
16. )
18. var _ db.DB = (*DB)(nil)
20. type DB struct {
21.         pebble   *pebble.DB
22.         wMutex   *sync.Mutex
23.         listener db.EventListener
24. }
26. // New opens a new database at the given path
27. func New(path string, cache uint, maxOpenFiles int, logger pebble.Logger) (db.DB, error) {
28.         // Ensure that the specified cache size meets a minimum threshold.
29.         cache = max(minCache, cache)
30.         pDB, err := newPebble(path, &pebble.Options{
31.                 Logger:       logger,
32.                 Cache:        pebble.NewCache(int64(cache * utils.Megabyte)),
33.                 MaxOpenFiles: maxOpenFiles,
34.         })
35.         if err != nil {
36.                 return nil, err
37.         }
38.         return pDB, nil
39. }
41. // NewMem opens a new in-memory database
42. func NewMem() (db.DB, error) {
43.         return newPebble("", &pebble.Options{
44.                 FS: vfs.NewMem(),
45.         })
46. }
48. // NewMemTest opens a new in-memory database, panics on error
49. func NewMemTest(t *testing.T) db.DB {
50.         memDB, err := NewMem()
51.         if err != nil {
52.                 t.Fatalf("create in-memory db: %v", err)
53.         }
54.         t.Cleanup(func() {
55.                 if err := memDB.Close(); err != nil {
56.                         t.Errorf("close in-memory db: %v", err)
57.                 }
58.         })
59.         return memDB
60. }
62. func newPebble(path string, options *pebble.Options) (*DB, error) {
63.         pDB, err := pebble.Open(path, options)
64.         if err != nil {
65.                 return nil, err
66.         }
67.         return &DB{pebble: pDB, wMutex: new(sync.Mutex), listener: &db.SelectiveListener{}}, nil
68. }
70. // WithListener registers an EventListener
71. func (d *DB) WithListener(listener db.EventListener) db.DB {
72.         d.listener = listener
73.         return d
74. }
76. // NewTransaction : see db.DB.NewTransaction
77. func (d *DB) NewTransaction(update bool) (db.Transaction, error) {
78.         txn := &Transaction{
79.                 listener: d.listener,
80.         }
81.         if update {
82.                 d.wMutex.Lock()
83.                 txn.lock = d.wMutex
84.                 txn.batch = d.pebble.NewIndexedBatch()
85.         } else {
86.                 txn.snapshot = d.pebble.NewSnapshot()
87.         }
89.         return txn, nil
90. }
92. // Close : see io.Closer.Close
93. func (d *DB) Close() error {
94.         return d.pebble.Close()
95. }
97. // View : see db.DB.View
98. func (d *DB) View(fn func(txn db.Transaction) error) error {
99.         return db.View(d, fn)
100. }
102. // Update : see db.DB.Update
103. func (d *DB) Update(fn func(txn db.Transaction) error) error {
104.         return db.Update(d, fn)
105. }
107. // Impl : see db.DB.Impl
108. func (d *DB) Impl() any {
109.         return d.pebble
110. }

复制代码

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

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

复制代码

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

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

1. func (d *DB) NewTransaction(update bool) (db.Transaction, error) {
2.         txn := &Transaction{
3.                 listener: d.listener,
4.         }
5.         if update {
6.                 d.wMutex.Lock()
7.                 txn.lock = d.wMutex
8.                 txn.batch = d.pebble.NewIndexedBatch()
9.         } else {
10.                 txn.snapshot = d.pebble.NewSnapshot()
11.         }
13.         return txn, nil
14. }

复制代码

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

   如许做的目的是为了包管数据库的一致性和隔离性。假如多个更新事件同时举行,可能会导致数据库状态出现不一致的情况。通过阻塞方式,可以确保同一时间只有一个更新事件在举行。
  db/pebble/transaction.go

1. package pebble
3. import (
4.         "errors"
5.         "io"
6.         "sync"
7.         "time"
9.         "github.com/NethermindEth/juno/db"
10.         "github.com/NethermindEth/juno/utils"
11.         "github.com/cockroachdb/pebble"
12. )
14. var ErrDiscardedTransaction = errors.New("discarded txn")
16. var _ db.Transaction = (*Transaction)(nil)
18. type Transaction struct {
19.         batch    *pebble.Batch
20.         snapshot *pebble.Snapshot
21.         lock     *sync.Mutex
22.         listener db.EventListener
23. }
25. // Discard : see db.Transaction.Discard
26. func (t *Transaction) Discard() error {
27.         if t.batch != nil {
28.                 if err := t.batch.Close(); err != nil {
29.                         return err
30.                 }
31.                 t.batch = nil
32.         }
33.         if t.snapshot != nil {
34.                 if err := t.snapshot.Close(); err != nil {
35.                         return err
36.                 }
37.                 t.snapshot = nil
38.         }
40.         if t.lock != nil {
41.                 t.lock.Unlock()
42.                 t.lock = nil
43.         }
44.         return nil
45. }
47. // Commit : see db.Transaction.Commit
48. func (t *Transaction) Commit() error {
49.         start := time.Now()
50.         defer func() { t.listener.OnCommit(time.Since(start)) }()
51.         if t.batch != nil {
52.                 return utils.RunAndWrapOnError(t.Discard, t.batch.Commit(pebble.Sync))
53.         }
54.         return utils.RunAndWrapOnError(t.Discard, ErrDiscardedTransaction)
55. }
57. // Set : see db.Transaction.Set
58. func (t *Transaction) Set(key, val []byte) error {
59.         start := time.Now()
60.         if t.batch == nil {
61.                 return errors.New("read only transaction")
62.         }
63.         if len(key) == 0 {
64.                 return errors.New("empty key")
65.         }
67.         defer func() { t.listener.OnIO(true, time.Since(start)) }()
68.         return t.batch.Set(key, val, pebble.Sync)
69. }
71. // Delete : see db.Transaction.Delete
72. func (t *Transaction) Delete(key []byte) error {
73.         start := time.Now()
74.         if t.batch == nil {
75.                 return errors.New("read only transaction")
76.         }
78.         defer func() { t.listener.OnIO(true, time.Since(start)) }()
79.         return t.batch.Delete(key, pebble.Sync)
80. }
82. // Get : see db.Transaction.Get
83. func (t *Transaction) Get(key []byte, cb func([]byte) error) error {
84.         start := time.Now()
85.         var val []byte
86.         var closer io.Closer
88.         var err error
89.         if t.batch != nil {
90.                 val, closer, err = t.batch.Get(key)
91.         } else if t.snapshot != nil {
92.                 val, closer, err = t.snapshot.Get(key)
93.         } else {
94.                 return ErrDiscardedTransaction
95.         }
97.         defer t.listener.OnIO(false, time.Since(start))
98.         if err != nil {
99.                 if errors.Is(err, pebble.ErrNotFound) {
100.                         return db.ErrKeyNotFound
101.                 }
103.                 return err
104.         }
105.         return utils.RunAndWrapOnError(closer.Close, cb(val))
106. }
108. // Impl : see db.Transaction.Impl
109. func (t *Transaction) Impl() any {
110.         if t.batch != nil {
111.                 return t.batch
112.         }
114.         if t.snapshot != nil {
115.                 return t.snapshot
116.         }
117.         return nil
118. }
120. // NewIterator : see db.Transaction.NewIterator
121. func (t *Transaction) NewIterator() (db.Iterator, error) {
122.         var (
123.                 iter *pebble.Iterator
124.                 err  error
125.         )
126.         if t.batch != nil {
127.                 iter, err = t.batch.NewIter(nil)
128.                 if err != nil {
129.                         return nil, err
130.                 }
131.         } else if t.snapshot != nil {
132.                 iter, err = t.snapshot.NewIter(nil)
133.                 if err != nil {
134.                         return nil, err
135.                 }
136.         } else {
137.                 return nil, ErrDiscardedTransaction
138.         }
140.         return &iterator{iter: iter}, nil
141. }

复制代码

pebble常用方法

NewSnapshot方法

NewSnapshot 方法，用于在 DB 类型上创建一个数据库的快照。
这个方法会返回一个 Snapshot 类型的对象，该对象代表了当前数据库状态的一个点时视图。
NewIndexedBatch 方法

NewIndexedBatch 方法，用于在 DB 类型上创建一个新的读写批处置惩罚对象。
参考

[视频、保举]数据存储与检索(详解b+树存储引擎(innodb、boltdb、buntdb等)、lsm树存储引擎(bitcask、moss、pebble、leveldb等))
参考URL: https://www.bilibili.com/video/BV1Zv411G7ty/

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。