• 《Go语言四十二章经》第四十章 LevelDB与BoltDB
    • 40.1 LevelDB
    • 40.2 BoltDB

    《Go语言四十二章经》第四十章 LevelDB与BoltDB

    作者：李骁

    LevelDB 和 BoltDB 都是k/v数据库。

    但LevelDB没有事务，LevelDB实现了一个日志结构化的merge tree。它将有序的key/value存储在不同文件的之中，通过db, _ := leveldb.OpenFile(“db”, nil)，在db目录下有很多数据文件，并通过“层级”把它们分开，并且周期性地将小的文件merge为更大的文件。这让其在随机写的时候会很快，但是读的时候却很慢。

    这也让LevelDB的性能不可预知：但数据量很小的时候，它可能性能很好，但是当随着数据量的增加，性能只会越来越糟糕。而且做merge的线程也会在服务器上出现问题。

    LSM树而且通过批量存储技术规避磁盘随机写入问题。 LSM树的设计思想非常朴素，它的原理是把一颗大树拆分成N棵小树， 它首先写入到内存中（内存没有寻道速度的问题，随机写的性能得到大幅提升），在内存中构建一颗有序小树，随着小树越来越大，内存的小树会flush到磁盘上。磁盘中的树定期可以做merge操作，合并成一棵大树，以优化读性能。

    BoltDB会在数据文件上获得一个文件锁，所以多个进程不能同时打开同一个数据库。BoltDB使用一个单独的内存映射的文件(.db)，实现一个写入时拷贝的B+树，这能让读取更快。而且，BoltDB的载入时间很快，特别是在从crash恢复的时候，因为它不需要去通过读log去找到上次成功的事务，它仅仅从两个B+树的根节点读取ID。

    BoltDB支持完全可序列化的ACID事务，让应用程序可以更简单的处理复杂操作。

    BoltDB设计源于LMDB，具有以下特点：

    • 直接使用API存取数据，没有查询语句；
    • 支持完全可序列化的ACID事务，这个特性比LevelDB强；
    • 数据保存在内存映射的文件里。没有wal、线程压缩和垃圾回收；
    • 通过COW技术，可实现无锁的读写并发，但是无法实现无锁的写写并发，这就注定了读性能超高，但写性能一般，适合与读多写少的场景。
    • 最后，BoltDB使用Golang开发，而且被应用于influxDB项目作为底层存储。

    LMDB的全称是Lightning Memory-Mapped Database(快如闪电的内存映射数据库)，它的文件结构简单，包含一个数据文件和一个锁文件.
    LMDB文件可以同时由多个进程打开，具有极高的数据存取速度，访问简单，不需要运行单独的数据库管理进程，只要在访问数据的代码里引用LMDB库，访问时给文件路径即可。

    让系统访问大量小文件的开销很大，而LMDB使用内存映射的方式访问文件，使得文件内寻址的开销非常小，使用指针运算就能实现。数据库单文件还能减少数据集复制/传输过程的开销。

    40.1 LevelDB

    1. package kvdb
    3. import (
    4.     "fmt"
    6.     "github.com/syndtr/goleveldb/leveldb"
    7.     "github.com/syndtr/goleveldb/leveldb/util"
    8. )
    10. func Leveldb() {
    11.     db, _ := leveldb.OpenFile("db", nil)
    13.     defer db.Close()
    14.     //读写数据库:
    15.     _ = db.Put([]byte("key1"), []byte("好好检查"), nil)
    16.     _ = db.Put([]byte("key2"), []byte("天天向上"), nil)
    17.     _ = db.Put([]byte("key:3"), []byte("就会一个本事"), nil)
    19.     data, _ := db.Get([]byte("key1"), nil)
    20.     fmt.Println(string(data))
    22.     //迭代数据库内容:
    23.     iter := db.NewIterator(nil, nil)
    24.     for iter.Next() {
    25.         key := iter.Key()
    26.         value := iter.Value()
    27.         fmt.Println(string(key), string(value))
    29.     }
    30.     iter.Release()
    31.     iter.Error()
    33.     //Seek-then-Iterate:
    34.     iter = db.NewIterator(nil, nil)
    35.     for ok := iter.Seek([]byte("key:")); ok; ok = iter.Next() {
    36.         // Use key/value.
    37.         fmt.Println("Seek-then-Iterate:")
    38.         fmt.Println(string(iter.Value()))
    39.     }
    40.     iter.Release()
    42.     //Iterate over subset of database content:
    43.     iter = db.NewIterator(&util.Range{Start: []byte("key:"), Limit: []byte("xoo")}, nil)
    44.     for iter.Next() {
    45.         // Use key/value.
    47.         fmt.Println("Iterate over subset of database content:")
    48.         fmt.Println(string(iter.Value()))
    49.     }
    50.     iter.Release()
    52.     //Iterate over subset of database content with a particular prefix:
    53.     iter = db.NewIterator(util.BytesPrefix([]byte("key")), nil)
    54.     for iter.Next() {
    55.         // Use key/value.
    57.         fmt.Println("Iterate over subset of database content with a particular prefix:")
    58.         fmt.Println(string(iter.Value()))
    59.     }
    60.     iter.Release()
    62.     _ = iter.Error()
    64.     //批量写:
    65.     batch := new(leveldb.Batch)
    66.     batch.Put([]byte("foo"), []byte("value"))
    67.     batch.Put([]byte("bar"), []byte("another value"))
    68.     batch.Delete([]byte("baz"))
    69.     _ = db.Write(batch, nil)
    71.     _ = db.Delete([]byte("key"), nil)
    72. }

    40.2 BoltDB

    1. package kvdb
    3. import (
    4.     "fmt"
    5.     "log"
    6.     "time"
    8.     "github.com/boltdb/bolt"
    9. )
    11. func Boltdb() error {
    12.     // Bolt 会在数据文件上获得一个文件锁，所以多个进程不能同时打开同一个数据库。
    13.     // 打开一个已经打开的 Bolt 数据库将导致它挂起，直到另一个进程关闭它。
    14.     // 为防止无限期等待，您可以将超时选项传递给Open()函数：
    15.     db, err := bolt.Open("my.db", 0600, &bolt.Options{Timeout: 1 * time.Second})
    16.     defer db.Close()
    17.     if err != nil {
    18.         log.Fatal(err)
    19.     }
    21.     //    两种处理方式：读-写和只读操作，读-写方式开始于db.Update方法：
    22.     //    Bolt 一次只允许一个读写事务，但是一次允许多个只读事务。 
    23. // 每个事务处理都有一个始终如一的数据视图
    24.     err = db.Update(func(tx *bolt.Tx) error {
    26.          // 这里还有另外一层：k-v存储在bucket中，
    27. // 可以将bucket当做一个key的集合或者是数据库中的表。
    28.          //（顺便提一句，buckets中可以包含其他的buckets，这将会相当有用）
    29.          // Buckets 是键值对在数据库中的集合.所有在bucket中的key必须唯一，
    30. // 使用DB.CreateBucket() 函数建立buket
    31.         //Tx.DeleteBucket() 删除bucket
    32.         //b := tx.Bucket([]byte("MyBucket"))
    33.         b, err := tx.CreateBucketIfNotExists([]byte("MyBucket"))
    36.         //要将 key/value 对保存到 bucket，请使用 Bucket.Put() 函数：
    37.         //这将在 MyBucket 的 bucket 中将 "answer" key的值设置为"42"。
    38.         err = b.Put([]byte("answer")，[]byte("42"))
    39.         return err
    40.     })
    42.     // 可以看到，传入db.update函数一个参数，在函数内部你可以get/set数据和处理error。
    43.     // 如返回为nil，事务就会从数据库得到一个commit，但如果返回一个实际的错误，则会做回滚，
    44.     // 你在函数中做的事情都不会commit。这很自然，因为你不需要人为地去关心事务的回滚，
    45.     // 只需要返回一个错误，其他的由Bolt去帮你完成。
    46.     // 只读事务 只读事务和读写事务不应该相互依赖，一般不应该在同一个例程中同时打开。
    47.     // 这可能会导致死锁，因为读写事务需要定期重新映射数据文件，
    48. // 但只有在只读事务处于打开状态时才能这样做。
    50.     // 批量读写事务.每一次新的事物都需要等待上一次事物的结束，
    51. // 可以通过DB.Batch()批处理来完
    52.     err = db.Batch(func(tx *bolt.Tx) error {
    54.         return nil
    55.     })
    57.     //只读事务在db.View函数之中：在函数中可以读取，但是不能做修改。
    58.     db.View(func(tx *bolt.Tx) error {
    59.         //要检索这个value，我们可以使用 Bucket.Get() 函数：
    60.         //由于Get是有安全保障的，所有不会返回错误，不存在的key返回nil
    61.         b := tx.Bucket([]byte("MyBucket"))
    62. //tx.Bucket([]byte("MyBucket")).Cursor() 可这样写
    63.         v := b.Get([]byte("answer"))
    64.         id, _ := b.NextSequence()
    65.         fmt.Printf("The answer is: %s %d \n", v, id)
    67.         //游标遍历key
    68.         c := b.Cursor()
    70.         for k, v := c.First(); k != nil; k, v = c.Next() {
    71.             fmt.Printf("key=%s, value=%s\n", k, v)
    72.         }
    74.         //游标上有以下函数：
    75.         //First()  移动到第一个健.
    76.         //Last()   移动到最后一个健.
    77.         //Seek()   移动到特定的一个健.
    78.         //Next()   移动到下一个健.
    79.         //Prev()   移动到上一个健.
    81.         //Prefix 前缀扫描
    82.         prefix := []byte("1234")
    83.         for k, v := c.Seek(prefix); k != nil && bytes.HasPrefix(k, prefix); k, v = c.Next() {
    84.             fmt.Printf("key=%s, value=%s\n", k, v)
    85.         }
    86.         return nil
    87.     })
    89.     //范围查找
    90.     //另一个常见的用例是扫描范围，例如时间范围。如果你使用一个合适的时间编码，如rfc3339然后可以查询特定日期范围的数据：
    91.     db.View(func(tx *bolt.Tx) error {
    92.         // Assume our events bucket exists and has RFC3339 encoded time keys.
    93.         c := tx.Bucket([]byte("Events")).Cursor()
    95.         // Our time range spans the 90's decade.
    96.         min := []byte("1990-01-01T00:00:00Z")
    97.         max := []byte("2000-01-01T00:00:00Z")
    99.         // Iterate over the 90's.
    100.         for k, v := c.Seek(min); k != nil && bytes.Compare(k, max) <= 0; k, v = c.Next() {
    101.             fmt.Printf("%s: %s\n", k, v)
    102.         }
    103.         return nil
    104.     })
    106.     //如果你知道所在桶中拥有键，你也可以使用ForEach()来迭代：
    107.     db.View(func(tx *bolt.Tx) error {
    108.         b := tx.Bucket([]byte("MyBucket"))
    110.         b.ForEach(func(k, v []byte) error {
    111.             fmt.Printf("key=%s, value=%s\n", k, v)
    112.             return nil
    113.         })
    114.         return nil
    115.     })
    117.     //事务处理
    118.     // 开始事务
    119.     tx, err := db.Begin(true)
    120.     if err != nil {
    121.         return err
    122.     }
    123.     defer tx.Rollback()
    125.     // 使用事务...
    126.     _, err = tx.CreateBucket([]byte("MyBucket"))
    127.     if err != nil {
    128.         return err
    129.     }
    131.     // 事务提交
    132.     if err = tx.Commit(); err != nil {
    133.         return err
    134.     }
    135.     return err
    137.     //还可以在一个键中存储一个桶，以创建嵌套的桶：
    138.     //func (*Bucket) CreateBucket(key []byte) (*Bucket, error)
    139.     //func (*Bucket) CreateBucketIfNotExists(key []byte) (*Bucket, error)
    140.     //func (*Bucket) DeleteBucket(key []byte) error
    141. }
    143. //备份 curl http://localhost/backup > my.db
    144. func BackupHandleFunc(w http.ResponseWriter, req *http.Request) {
    145.     err := db.View(func(tx *bolt.Tx) error {
    146.         w.Header().Set("Content-Type", "application/octet-stream")
    147.         w.Header().Set("Content-Disposition", `attachment; filename="my.db"`)
    148.         w.Header().Set("Content-Length", strconv.Itoa(int(tx.Size())))
    149.         _, err := tx.WriteTo(w)
    150.         return err
    151.     })
    152.     if err != nil {
    153.         http.Error(w, err.Error(), http.StatusInternalServerError)
    154.     }
    155. }
    157. //桶的自增
    158. //利用nextsequence()功能，你可以让boltdb生成序列作为你键值对的唯一标识。见下面的示例。
    159. func (s *Store) CreateUser(u *User) error {
    160.     return s.db.Update(func(tx *bolt.Tx) error {
    161.         // 创建users桶
    162.         b := tx.Bucket([]byte("users"))
    164.         // 生成自增序列
    165.         id, _ = b.NextSequence()
    166.         u.ID = int(id)
    168.         // Marshal user data into bytes.
    169.         buf, err := Json.Marshal(u)
    170.         if err != nil {
    171.             return err
    172.         }
    173.         // Persist bytes to users bucket.
    174.         return b.Put(itob(u.ID), buf)
    175.     })
    176. }
    178. // itob returns an 8-byte big endian representation of v.
    179. func itob(v int) []byte {
    180.     b := make([]byte, 8)
    181.     binary.BigEndian.PutUint64(b, uint64(v))
    182.     return b
    183. }
    185. type User struct {
    186.     ID int
    187. }