SSDB源码分析（2）：SSDB

    SSDB类封装了对存储的各种操作，上文的结尾看到服务进程启动时会打开存储系统数据文件和元数据文件，本文将深入介绍SSDB类和其派生类SSDBImpl。
    SSDB是一个抽象类，它除了定义对存储进行操作的各个接口，还提供了打开数据库文件的方法SSDB::open。
    

class SSDB{
public:
    SSDB(){}
    virtual ~SSDB(){};
    static SSDB* open(const Options &opt, const std::string &base_dir);

    // return (start, end], not include start
    virtual Iterator* iterator(const std::string &start, const std::string &end, uint64_t limit) = 0;
    virtual Iterator* rev_iterator(const std::string &start, const std::string &end, uint64_t limit) = 0;

    //void flushdb();
    virtual uint64_t size() = 0;
    virtual std::vector<std::string> info() = 0;
    virtual void compact() = 0;
    virtual int key_range(std::vector<std::string> *keys) = 0;

    /* raw operates */

    // repl: whether to sync this operation to slaves
    virtual int raw_set(const Bytes &key, const Bytes &val) = 0;
    virtual int raw_del(const Bytes &key) = 0;
    virtual int raw_get(const Bytes &key, std::string *val) = 0;

    /* key value */

    virtual int set(const Bytes &key, const Bytes &val, char log_type=BinlogType::SYNC) = 0;
    virtual int setnx(const Bytes &key, const Bytes &val, char log_type=BinlogType::SYNC) = 0;
    virtual int del(const Bytes &key, char log_type=BinlogType::SYNC) = 0;
    // -1: error, 1: ok, 0: value is not an integer or out of range
    virtual int incr(const Bytes &key, int64_t by, int64_t *new_val, char log_type=BinlogType::SYNC) = 0;
    virtual int multi_set(const std::vector<Bytes> &kvs, int offset=0, char log_type=BinlogType::SYNC) = 0;
    virtual int multi_del(const std::vector<Bytes> &keys, int offset=0, char log_type=BinlogType::SYNC) = 0;
    virtual int setbit(const Bytes &key, int bitoffset, int on, char log_type=BinlogType::SYNC) = 0;
    virtual int getbit(const Bytes &key, int bitoffset) = 0;

    virtual int get(const Bytes &key, std::string *val) = 0;
    virtual int getset(const Bytes &key, std::string *val, const Bytes &newval, char log_type=BinlogType::SYNC) = 0;
    // return (start, end]
    virtual KIterator* scan(const Bytes &start, const Bytes &end, uint64_t limit) = 0;
    virtual KIterator* rscan(const Bytes &start, const Bytes &end, uint64_t limit) = 0;

    /* hash */

    virtual int hset(const Bytes &name, const Bytes &key, const Bytes &val, char log_type=BinlogType::SYNC) = 0;
    virtual int hdel(const Bytes &name, const Bytes &key, char log_type=BinlogType::SYNC) = 0;
    // -1: error, 1: ok, 0: value is not an integer or out of range
    virtual int hincr(const Bytes &name, const Bytes &key, int64_t by, int64_t *new_val, char log_type=BinlogType::SYNC) = 0;

    virtual int64_t hsize(const Bytes &name) = 0;
    virtual int64_t hclear(const Bytes &name) = 0;
    virtual int hget(const Bytes &name, const Bytes &key, std::string *val) = 0;
    virtual int hlist(const Bytes &name_s, const Bytes &name_e, uint64_t limit,
            std::vector<std::string> *list) = 0;
    virtual int hrlist(const Bytes &name_s, const Bytes &name_e, uint64_t limit,
            std::vector<std::string> *list) = 0;
    virtual HIterator* hscan(const Bytes &name, const Bytes &start, const Bytes &end, uint64_t limit) = 0;
    virtual HIterator* hrscan(const Bytes &name, const Bytes &start, const Bytes &end, uint64_t limit) = 0;

    /* zset */

    virtual int zset(const Bytes &name, const Bytes &key, const Bytes &score, char log_type=BinlogType::SYNC) = 0;
    virtual int zdel(const Bytes &name, const Bytes &key, char log_type=BinlogType::SYNC) = 0;
    // -1: error, 1: ok, 0: value is not an integer or out of range
    virtual int zincr(const Bytes &name, const Bytes &key, int64_t by, int64_t *new_val, char log_type=BinlogType::SYNC) = 0;

    virtual int64_t zsize(const Bytes &name) = 0;
    /**
     * @return -1: error; 0: not found; 1: found
     */
    virtual int zget(const Bytes &name, const Bytes &key, std::string *score) = 0;
    virtual int64_t zrank(const Bytes &name, const Bytes &key) = 0;
    virtual int64_t zrrank(const Bytes &name, const Bytes &key) = 0;
    virtual ZIterator* zrange(const Bytes &name, uint64_t offset, uint64_t limit) = 0;
    virtual ZIterator* zrrange(const Bytes &name, uint64_t offset, uint64_t limit) = 0;
    /**
     * scan by score, but won't return @key if key.score=score_start.
     * return (score_start, score_end]
     */
    virtual ZIterator* zscan(const Bytes &name, const Bytes &key,
            const Bytes &score_start, const Bytes &score_end, uint64_t limit) = 0;
    virtual ZIterator* zrscan(const Bytes &name, const Bytes &key,
            const Bytes &score_start, const Bytes &score_end, uint64_t limit) = 0;
    virtual int zlist(const Bytes &name_s, const Bytes &name_e, uint64_t limit,
            std::vector<std::string> *list) = 0;
    virtual int zrlist(const Bytes &name_s, const Bytes &name_e, uint64_t limit,
            std::vector<std::string> *list) = 0;

    virtual int64_t qsize(const Bytes &name) = 0;
    // @return 0: empty queue, 1: item peeked, -1: error
    virtual int qfront(const Bytes &name, std::string *item) = 0;
    // @return 0: empty queue, 1: item peeked, -1: error
    virtual int qback(const Bytes &name, std::string *item) = 0;
    // @return -1: error, other: the new length of the queue
    virtual int64_t qpush_front(const Bytes &name, const Bytes &item, char log_type=BinlogType::SYNC) = 0;
    virtual int64_t qpush_back(const Bytes &name, const Bytes &item, char log_type=BinlogType::SYNC) = 0;
    // @return 0: empty queue, 1: item popped, -1: error
    virtual int qpop_front(const Bytes &name, std::string *item, char log_type=BinlogType::SYNC) = 0;
    virtual int qpop_back(const Bytes &name, std::string *item, char log_type=BinlogType::SYNC) = 0;
    virtual int qfix(const Bytes &name) = 0;
    virtual int qlist(const Bytes &name_s, const Bytes &name_e, uint64_t limit,
            std::vector<std::string> *list) = 0;
    virtual int qrlist(const Bytes &name_s, const Bytes &name_e, uint64_t limit,
            std::vector<std::string> *list) = 0;
    virtual int qslice(const Bytes &name, int64_t offset, int64_t limit,
            std::vector<std::string> *list) = 0;
    virtual int qget(const Bytes &name, int64_t index, std::string *item) = 0;
    virtual int qset(const Bytes &name, int64_t index, const Bytes &item, char log_type=BinlogType::SYNC) = 0;
    virtual int qset_by_seq(const Bytes &name, uint64_t seq, const Bytes &item, char log_type=BinlogType::SYNC) = 0;
};

SSDB::open

    SSDB::open函数根据配置参数和路径打开数据库文件，由于SSDB是基于leveldb的，所以实际上是调用leveldb的接口创建和打开数据库。

SSDB* SSDB::open(const Options &opt, const std::string &dir){
    SSDBImpl *ssdb = new SSDBImpl();
    // 如果数据库不存在则创建
    ssdb->options.create_if_missing = true;

    // 打开的.sst文件fd的最大数目
    ssdb->options.max_open_files = opt.max_open_files;

    // LevelDB增加了bloomfilter支持, 可以过滤掉一些key不存在的情况, 减少对磁盘的访问
    ssdb->options.filter_policy = leveldb::NewBloomFilterPolicy(10);

    // block cache 缓存的block数据，block是sstable文件内组织数据的单位，也是从磁盘中读取和写入的单位
    // block默认大小为4KB，可以使用options.block_size设置，最小1KB，最大4MB。对于频繁做scan操作的应用，
    // 可适当调大此参数，对大量小value随机读取的应用，也可尝试调小该参数
    ssdb->options.block_cache = leveldb::NewLRUCache(opt.cache_size * 1048576);

    ssdb->options.block_size = opt.block_size * 1024;
    ssdb->options.write_buffer_size = opt.write_buffer_size * 1024 * 1024;
    ssdb->options.compaction_speed = opt.compaction_speed;
    if(opt.compression == "yes"){
        ssdb->options.compression = leveldb::kSnappyCompression;
    }else{
        ssdb->options.compression = leveldb::kNoCompression;
    }

    leveldb::Status status;
    // 打开leveldb数据库
    status = leveldb::DB::Open(ssdb->options, dir, &ssdb->db);
    if(!status.ok()){
        log_error("open db failed");
        goto err;
    }
    // 创建操作的日志队列，用于主从备份
    ssdb->binlogs = new BinlogQueue(ssdb->db, opt.binlog);

    return ssdb;
err:
    if(ssdb){
        delete ssdb;
    }
    return NULL;
}

open函数主要完成三个任务:
    1. 创建对应的SSDBImple对象;
    2. 利用传入的配置参数打开leveldb数据库作为存储(SSDB的数据文件和元数据文件各使用一个leveldb数据库？对这里还有疑问，MyApplication::run函数中的meta_db到底起到什么作用);
    3. 创建操作日志队列，用于主从同步时将这些操作在从服务器上执行一遍，从而备份数据到从服务器上;