LevelDB 代碼擼起來！

LevelDB 的大致原理已經講完了，本節我們要親自使用 Java 語言第三方庫 leveldbjni 來實踐一下 LevelDB 的各種特性。這個庫使用了 Java Native Interface 計數將 C++ 實現的 LevelDB 包裝成了 Java 平臺 的 API。其它語言同樣也是采用了類似 JNI 的技術來包裝的 LevelDB。

Maven 依賴

下載下面的依賴包地址，你就可以得到一個支持全平臺的 jar 包。LevelDB 在不同的操作系統平臺會編譯出不同的動態鏈接庫形式，這個 jar 包將所有平臺的動態鏈接庫都包含進來了。

<dependencies>
  <dependency>
    <groupId>org.fusesource.leveldbjni</groupId>
    <artifactId>leveldbjni-linux64</artifactId>
    <version>1.8</version>
  </dependency>
</dependencies>

增查刪 API

這個例子中我們將自動創建一個 LevelDB 數據庫，然后往里面塞入 100w 條數據，再取出來，再刪掉所有數據。這個例子在我的 Mac 上會運行了大約 10s 的時間。也就是說讀寫平均 QPS 高達 30w/s，完全可以媲美 Redis，不過這大概也是因為鍵值對都比較小，在實際生產環境中性能應該沒有這么高，它至少應該比 Redis 稍慢一些。

import static org.fusesource.leveldbjni.JniDBFactory.factory;

import java.io.File;
import java.io.IOException;

import org.iq80.leveldb.DB;
import org.iq80.leveldb.Options;

public class Sample {

    public static void main(String[] args) throws IOException {
        Options options = new Options();
        options.createIfMissing(true);
        DB db = factory.open(new File("/tmp/lvltest"), options);
        try {
            for (int i = 0; i < 1000000; i++) {
                byte[] key = new String("key" + i).getBytes();
                byte[] value = new String("value" + i).getBytes();
                db.put(key, value);
            }
            for (int i = 0; i < 1000000; i++) {
                byte[] key = new String("key" + i).getBytes();
                byte[] value = db.get(key);
                String targetValue = "value" + i;
                if (!new String(value).equals(targetValue)) {
                    System.out.println("something wrong!");
                }
            }
            for (int i = 0; i < 1000000; i++) {
                byte[] key = new String("key" + i).getBytes();
                db.delete(key);
            }
        } finally {
            db.close();
        }
    }
}

我們再觀察數據庫的目錄中，LevelDB 都創建了那些東西

這個目錄里我們看到了有很多 sst 擴展名的文件，它就是 LevelDB 的磁盤數據文件，有些 LevelDB 的版本數據文件的擴展名是 ldb，不過內部格式一樣，僅僅是擴展名不同而已。其中還有一個 log 擴展名的文件，這就是操作日志文件，記錄了最近一段時間的操作日志。其它幾個大些名稱文件我們先不必去了解，后續我們再仔細解釋。

將這個目錄里面的文件全部刪掉，這個庫就徹底清空了。

也許你會想到上面的例子中不是所有的數據最終都被刪除了么，怎么還會有如此多的 sst 數據文件呢？這是因為 LevelDB 的刪除操作并不是真的立即刪除鍵值對，而是將刪除操作轉換成了更新操作寫進去了一個特殊的鍵值對，這個鍵值對的值部分是一個特殊的刪除標記。

待 LevelDB 在某種條件下觸發數據合并（compact）時才會真的刪除相應的鍵值對。

數據合并

LevelDB 提供了數據合并的手動調用 API，下面我們手動整理一下，看看整理后會發生什么

public void compactRange(byte[] begin, byte[] end)

這個 API 可以選擇范圍進行整理，如果 begin 參數為 null，那就表示從頭開始，如果 end 參數為 null，那就表示一直到尾部。

public static void main(String[] args) throws IOException {
    Options options = new Options();
    options.createIfMissing(true);
    DB db = factory.open(new File("/tmp/lvltest"), options);
    try {
        db.compactRange(null, null);
    } finally {
        db.close();
    }
}

運行了大約 1s 多點時間，完畢后我們看到目錄中 sst 文件沒有了

如果我們沒有手工調用數據整理 API，LevelDB 內部也有一定的策略來定期整理。

讀性能

如果我們將上面的代碼加上時間打點，觀察讀寫性能差異，你會發現一個有趣的現象，那就是寫性能比讀性能還要好，雖然本例中所有的讀操作都是命中的。

put: 3150ms
get: 4128ms
delete: 1983ms

這是因為寫操作記完操作日志將數據寫進內存后就返回了，而讀操作有可能內存讀 miss，然后要去磁盤讀。之所以讀寫性能差距不是非常明顯，是因為 LevelDB 會緩存最近一次讀取的數據塊，而且我的個人電腦的磁盤是 SSD 磁盤，讀性能都好。如果是普通磁盤，就會看出明顯的性能差異了。

下面我們將讀操作改成隨機讀，就會發現讀寫性能發生很大的差別

for (int i = 0; i < 1000000; i++) {
    int index = ThreadLocalRandom.current().nextInt(1000000);
    byte[] key = new String("key" + index).getBytes();
    db.get(key);
}

--------
put: 3094ms
get: 9781ms
delete: 1969ms

這時要改善讀性能就可以借助塊緩存了

// 設置 100M 的塊緩存
options.cacheSize(100 * 1024 * 1024);

------------
put: 2877ms
get: 4758ms
delete: 1981ms

同步 vs 異步

上一節我們提到 LevelDB 還提供了同步寫的 API，確保操作日志落地后才 put 方法才返回。它的性能會明顯弱于普通寫操作，下面我們來對比一下兩者的性能差異。

public static void main(String[] args) throws IOException {
    long start = System.currentTimeMillis();
    Options options = new Options();
    options.createIfMissing(true);
    DB db = factory.open(new File("/tmp/lvltest"), options);
    try {
        for (int i = 0; i < 1000000; i++) {
            byte[] key = new String("key" + i).getBytes();
            byte[] value = new String("value" + i).getBytes();
            WriteOptions wo = new WriteOptions();
            wo.sync(true);
            db.put(key, value, wo);
        }
    } finally {
        db.close();
    }
    long end = System.currentTimeMillis();
    System.out.println(end - start);
}

上面這個同步寫操作足足花了 90s 多的時間。將 sync 選項去掉后，只需要 3s 多點。性能差距高達 30 倍。下面我們來簡單改造一下上面的代碼，讓它變成間隔同步寫，也就是每隔 N 個寫操作同步一次，取 N = 100。

WriteOptions wo = new WriteOptions();
wo.sync(i % 100 == 0);

運行時間變成了不到 5s。再將 N 改成 10，運行時間變成了不到 12s。即使是 12s，寫的平均 QPS 也高達 8w/s，這還是很客觀的。

普通寫 VS 批量寫

LevelDB 提供了批量寫操作，它會不會類似于 Redis 的管道可以加快指令的運行呢，下面我們來嘗試使用 WriteBatch，對比一下普通的寫操作，看看性能差距有多大。

public static void main(String[] args) throws IOException {
    long start = System.currentTimeMillis();
    Options options = new Options();
    options.createIfMissing(true);
    DB db = factory.open(new File("/tmp/lvltest"), options);
    try {
        WriteBatch batch = db.createWriteBatch();
        for (int i = 0; i < 1000000; i++) {
            byte[] key = new String("key" + i).getBytes();
            byte[] value = new String("value" + i).getBytes();
            batch.put(key, value);
            if (i % 100 == 0) {
                db.write(batch);
                batch.close();
                batch = db.createWriteBatch();
            }
        }
        db.write(batch);
        batch.close();
    } finally {
        db.close();
    }

 long end = System.currentTimeMillis();
    System.out.println(end - start);
}

將批次數量 N 分別改成 10、100、1000，運行后可以發現耗時差不多，大約都是 2s 多點。這意味著批量寫并不會大幅提升寫操作的吞吐量。但是將 N 改成 1 后你會發現耗時和普通寫操作相差無幾，大約是 3s 多，再將 N 改成 2、5 等，耗時還是會有所降低，到 2s 多 左右就穩定了，此時提升 N 值不再有明顯效果。這意味著批量寫操作確實會比普通寫快一點，但是相差也不會過大。它不同于 Redis 的管道可以大幅減少網絡開銷帶來的明顯性能提升，LevelDB 是純內存數據庫，根本談不上網絡開銷。

那為什么批量寫還是會比普通寫快一點呢？要回答這個問題就需要追蹤 LevelDB 的源碼，還在這部分邏輯比較簡單，大家應該都可以理解，所以這里就直接貼出來了。

Status DB::Put(WriteOptions& opt, Slice& key, Slice& value) {
  WriteBatch batch;
  batch.Put(key, value);
  return Write(opt, &batch);
}

很明顯，每一個普通寫操作最終都會被轉換成一個批量寫操作，只不過 N=1 。這正好解釋了為什么當 N=1 時批量寫操作和普通寫操作相差無幾。

我們再繼續追蹤 WriteBatch 的源碼我發現每一個批量寫操作都需要使用互斥鎖。當批次 N 值比較大時，相當于加鎖的平均次數減少了，于是整體性能就提升了。但是也不會提升太多，因為加鎖本身的損耗占比開銷也不是特別大。這也意味著在多線程場合，寫操作性能會下降，因為鎖之間的競爭將導致內耗增加。

為什么說批量寫可以保證內部一系列操作的原子性呢，就是因為這個互斥鎖的保護讓寫操作單線程化了。因為這個粗粒度鎖的存在，LevelDB 寫操作的性能被大大限制了。這也成了后來居上的 RocksDB 重點優化的方向。

快照和遍歷

LevelDB 提供了快照讀功能可以保證同一個快照內同一個 Key 讀到的數據保持一致，避免「不可重復讀」的發生。下面我們使用快照來嘗試一下遍歷操作，在遍歷的過程中順便還修改對應 Key 的值，看看快照讀是否可以隔離寫操作。

public static void main(String[] args) throws IOException {
    Options options = new Options();
    options.createIfMissing(true);
    DB db = factory.open(new File("/tmp/lvltest"), options);
    try {
        for (int i = 0; i < 10000; i++) {
            String padding = String.format("%04d", i);
            byte[] key = new String("key" + padding).getBytes();
            byte[] value = new String("value" + padding).getBytes();
            db.put(key, value);
        }

        Snapshot ss = db.getSnapshot();
        // 掃描
        scan(db, ss);
        // 修改
        for (int i = 0; i < 10000; i++) {
            String padding = String.format("%04d", i);
            byte[] key = new String("key" + padding).getBytes();
            byte[] value = new String("!value" + padding).getBytes(); // 修改
            db.put(key, value);
        }
        // 再掃描
        scan(db, ss);
        ss.close();
    } finally {
        db.close();
    }
}

private static void scan(DB db, Snapshot ss) throws IOException {
    ReadOptions ro = new ReadOptions();
    ro.snapshot(ss);
    DBIterator it = db.iterator(ro);
    int k = 0;
    // it.seek(someKey); // 從指定位置開始遍歷
    it.seekToFirst(); // 從頭開始遍歷
    while (it.hasNext()) {
        Entry<byte[], byte[]> entry = it.next();
        String key = new String(entry.getKey());
        String value = new String(entry.getValue());
        String padding = String.format("%04d", k);
        String targetKey = new String("key" + padding);
        String targetVal = new String("value" + padding);
        if (!targetKey.equals(key) || !targetVal.equals(value)) {
            System.out.printf("something wrong");
        }
        k++;
    }

    System.out.printf("total %d\n", k);
    it.close();
}
--------------------
total 10000
total 10000

前后兩次遍歷從快照中獲取到的數據還是一致的，也就是說中間的寫操作根本沒有影響到快照的狀態，這就是我們想要的結果。那快照的原理是什么呢？

快照的原理其實非常簡單，簡單到讓人懷疑人生。對于庫中的每一個鍵值對，它會因為修改操作而存在多個值的版本。在數據庫文件內容合并之前，同一個 Key 可能會存在于多個文件中，每個文件中的值版本不一樣。這個版本號是由數據庫唯一的全局自增計數值標記的。快照會記錄當前的計數值，在當前快照里讀取的數據都需要和快照的計數值比對，只有小于這個計數值才是有效的數據版本。

既然同一個 Key 存在多個版本的數據，對于同一個 Key，遍歷操作是如何避免重復的呢？關于這個問題我們后續再深入探討。

布隆過濾器

leveldbjni 沒有封裝 LevelDB 提供的布隆過濾器功能。所以為了嘗試布隆過濾器的效果，我們需要試試其它語言，這里我使用 Go 語言的 levigo 庫。

// 安裝 leveldb和snappy庫
$ brew install leveldb
// 再安裝 levigo
$ go get github.com/jmhodges/levigo

這個例子中我們將寫入更多的數據 —— 1000w 條，當數據量增多時，LevelDB 將形成更深的層級。同時為了構造出讀 miss 的效果，我們寫入偶數的鍵值對，然后再隨機讀取奇數的鍵值對。再對比增加布隆過濾器前后的讀性能差異。

package main

import (
    "fmt"
    "math/rand"
    "time"

    "github.com/jmhodges/levigo"
)

func main() {
    options := levigo.NewOptions()
    options.SetCreateIfMissing(true)
    // 每個 key 占用 10個bit
    // options.SetFilterPolicy(levigo.NewBloomFilter(10))

    db, _ := levigo.Open("/tmp/lvltest", options)
    start := time.Now().UnixNano()
    for i := 0; i < 10000000; i++ {
        key := []byte(fmt.Sprintf("key%d", i*2))
        value := []byte(fmt.Sprintf("value%d", i*2))
        wo := levigo.NewWriteOptions()
        db.Put(wo, key, value)
    }
    duration := time.Now().UnixNano() - start
    fmt.Println("put:", duration/1e6, "ms")
    start = time.Now().UnixNano()
    for i := 0; i < 10000000; i++ {
        index := rand.Intn(10000000)
        key := []byte(fmt.Sprintf("key%d", index*2+1))
        ro := levigo.NewReadOptions()
        db.Get(ro, key)
    }
    duration = time.Now().UnixNano() - start
    fmt.Println("get:", duration/1e6, "ms")
    start = time.Now().UnixNano()
    for i := 0; i < 10000000; i++ {
        key := []byte(fmt.Sprintf("key%d", i*2))
        wo := levigo.NewWriteOptions()
        db.Delete(wo, key)
    }
    duration = time.Now().UnixNano() - start
    fmt.Println("get:", duration/1e6, "ms")
}

-----------
put: 61054ms
get: 104942ms
get: 47269ms

再去掉注釋，打開布隆過濾器，觀察結果

put: 57653ms
get: 36895ms
get: 57554ms

可以明顯看出，讀性能提升了 3 倍，這是一個非常了不起的性能提升。在讀 miss 開啟了布隆過濾器的情況下，我們再試試打開塊緩存，看看是否還能再繼續提升讀性能

put: 57022ms
get: 37475ms
get: 58999ms

結論是在讀 miss 開啟了布隆過濾器場景下塊緩存幾乎不起作用。但是這并不是說塊緩存沒有用，在讀命中的情況下，塊緩存的作用還是很大的。

布隆過濾器在顯著提升性能的同時，也是需要浪費一定的磁盤空間。LevelDB 需要將布隆過濾器的二進制數據存儲到數據塊中，不過布隆過濾器的空間占比相對而言不是很高，完全在可接受范圍之內。

壓縮

LevelDB 的壓縮算法采用 Snappy，這個算法解壓縮效率很高，在壓縮比相差不大的情況下 CPU 消耗很低。官方不建議關閉壓縮算法，不過經過我的測試發現，關閉壓縮確實可以顯著提升讀性能。不過關閉了壓縮，這也意味著你的磁盤空間要浪費好幾倍，這代價也不低。

public static void main(String[] args) throws IOException {
    Options options = new Options();
    options.createIfMissing(true);
    options.compressionType(CompressionType.None);
    DB db = factory.open(new File("/tmp/lvltest"), options);
    try {
        long start = System.currentTimeMillis();
        for (int i = 0; i < 1000000; i++) {
            byte[] key = new String("key" + 2 * i).getBytes();
            byte[] value = new String("value" + 2 * i).getBytes();
            db.put(key, value);
        }
        long duration = System.currentTimeMillis() - start;
        System.out.printf("put:%dms\n", duration);
        start = System.currentTimeMillis();
        for (int i = 0; i < 1000000; i++) {
            int index = ThreadLocalRandom.current().nextInt(1000000);
            byte[] key = new String("key" + (2 * index + 1)).getBytes();
            db.get(key);
        }
        duration = System.currentTimeMillis() - start;
        System.out.printf("get:%dms\n", duration);
        start = System.currentTimeMillis();
        for (int i = 0; i < 1000000; i++) {
            byte[] key = new String("key" + 2 * i).getBytes();
            db.delete(key);
        }
        duration = System.currentTimeMillis() - start;
        System.out.printf("delete:%dms\n", duration);
    } finally {
        db.close();
    }
}

----------------
put:3785ms
get:6475ms
delete:1935ms

下面我們再打開壓縮，對比一下結果，讀性能差距接近 1 倍

options.compressionType(CompressionType.SNAPPY);

---------------
put:3804ms
get:11644ms
delete:2750m

下一節將開始深入 LevelDB 實現原理，先從 LevelDB 的宏觀結構開