SSDB --- queue实现

SSDB介绍

SSDB是一款nosql数据库，使用leveldb作为存储引擎，支持众多的数据结构，如 queue、zset和hash，宣称可用来替代redis，看来性能不可小觑。其作者是位国人，曾在360游戏部门任职过。SSDB被很多知名公司使用，代码质量不错。

http://ssdb.io/zh_cn/，代码托管在github。

queue的实现

leveldb本身是k/v数据库，如何在单纯的k/v上实现queue呢？

来看ssdb中的实现：

下面是ssdb/src/ssdb/ssdb.h 中queue部分的接口

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/src/ssdb/t_queue.cpp中。

队列只能在头和尾插入或删除，所以先来看qpush_front函数：

int64_t SSDBImpl::qpush_front(const Bytes &name, const Bytes &item, char log_type){
return _qpush(name, item, QFRONT_SEQ, log_type);
//QFRONT_SEQ 指示从前头插入
}

调用了_qpush():

int64_t SSDBImpl::_qpush(const Bytes &name, const Bytes &item, uint64_t front_or_back_seq, char log_type){
Transaction trans(binlogs);//

int ret;
// generate seq
uint64_t seq;
ret = qget_uint64(this->db, name,  front_or_back_seq, &seq); //根据当前头或者尾生成合适的seq号
if(ret == -1){
return -1;
}
// update front and/or back
//下面的这个if else是整个queue实现的核心，在leveldb中维护 queue的头和尾的 seq。
if(ret == 0){ //如果队列里没元素
seq = QITEM_SEQ_INIT;
//const uint64_t QITEM_SEQ_INIT = QITEM_MAX_SEQ/2;
// const uint64_t QITEM_MAX_SEQ = 9223372036854775807ULL;这是个天文数字，以seq构造队列
// 效果是 min-------------中位数---------------max，然后从中位数seq向两边扩大 就构成了一个queue
ret = qset_one(this, name, QFRONT_SEQ, Bytes(&seq, sizeof(seq)));  //将中位数seq当作头seq写入数据库
if(ret == -1){
return -1;
}
ret = qset_one(this, name, QBACK_SEQ, Bytes(&seq, sizeof(seq))); //将中位数seq当作尾seq写入数据库，现在头，尾指向一样的seq。
}else{
seq += (front_or_back_seq == QFRONT_SEQ)? -1 : +1;//如果是在头部插入，则把seq - 1，否则 seq + 1
ret = qset_one(this, name, front_or_back_seq, Bytes(&seq, sizeof(seq))); 更新数据库中头或尾seq
}
if(ret == -1){
return -1;
}
if(seq <= QITEM_MIN_SEQ || seq >= QITEM_MAX_SEQ){ //队列满了
log_info("queue is full, seq: %" PRIu64 " out of range", seq);
return -1;
}

// prepend/append item
ret = qset_one(this, name, seq, item);//在当前seq处插入 节点。
if(ret == -1){
return -1;
}

std::string buf = encode_qitem_key(name, seq);
if(front_or_back_seq == QFRONT_SEQ){
binlogs->add_log(log_type,  BinlogCommand::QPUSH_FRONT, buf); //提交leveldb的writebatch (批量写)
}else{
binlogs->add_log(log_type, BinlogCommand::QPUSH_BACK, buf);
}

// update size
int64_t size = incr_qsize(this, name, +1);//queue的size +1
if(size == -1){
return -1;
}

leveldb::Status s = binlogs->commit();//提交给leveldb写入
if(!s.ok()){
log_error("Write error!");
return -1;
}
return size;
}

qset_one:

static int qset_one(SSDBImpl *ssdb, const Bytes &name, uint64_t seq, const Bytes &item){
std::string key = encode_qitem_key(name, seq);
leveldb::Status s;

ssdb->binlogs->Put(key, slice(item));
return 0;
}

这个函数其实是将seq 和item 编码成一个key，然后再提交leveldb 的writebatch(批量写)。

pop操作与push基本相同，就不分析了。