有序集合是Redis對象系統(tǒng)中的一部分，其底層采用跳表和壓縮列表兩種形式存儲，在上一篇介紹了跳表實(shí)現(xiàn)，就趁熱打鐵看一下有序集合的跳表實(shí)現(xiàn)

本篇主要涉及的是有序集合添加數(shù)據(jù)的命令，后面會看到，在命令的底層實(shí)現(xiàn)中，實(shí)際上還是調(diào)用跳表的接口

存儲結(jié)構(gòu)

有序集合的定義在server.h文件中，不過除了跳表以外，有序集合又保存了一個字典，這個字典的作用是用來查找某個數(shù)據(jù)對應(yīng)的分值。根據(jù)跳表的實(shí)現(xiàn)可知，跳表內(nèi)部是采用分值排序的，通過分值查找數(shù)據(jù)還行，但是如果要通過數(shù)據(jù)查找分值，就顯得力不從心了，所以Redis又維護(hù)了一個字典，用來完成通過數(shù)據(jù)查找分值的任務(wù)

//server.h /* 有序集合 */ typedef struct zset {dict *dict; /* 存儲<數(shù)據(jù)，分值>鍵值對，用來通過數(shù)據(jù)查找分值 */zskiplist *zsl; /* 跳表，保存<分值，數(shù)據(jù)>，內(nèi)部通過分值排序 */ } zset;

有序集合操作

添加數(shù)據(jù)

通過命令ZADD，可以實(shí)現(xiàn)向數(shù)據(jù)庫中添加有序集合，該命令是由zaddGenericCommand函數(shù)實(shí)現(xiàn)的。函數(shù)中會先解析命令選項(xiàng)，命令參數(shù)，然后根據(jù)底層是由跳表實(shí)現(xiàn)還是由壓縮列表實(shí)現(xiàn)執(zhí)行不同的操作，都是調(diào)用二者的接口

/* 向有序集合中添加數(shù)據(jù) */ void zaddGenericCommand(client *c, int flags) {static char *nanerr = "resulting score is not a number (NaN)";/* 獲取鍵 */robj *key = c->argv[1];robj *ele;robj *zobj;robj *curobj;double score = 0, *scores = NULL, curscore = 0.0;int j, elements;int scoreidx = 0;...scoreidx = 2;/* 獲取參數(shù)選項(xiàng)，參數(shù)選項(xiàng)緊接在鍵的后面 */.../* 將參數(shù)選項(xiàng)轉(zhuǎn)換為數(shù)值變量 */.../* argc中保存所有參數(shù)個數(shù)，scoreidx保存第一個數(shù)據(jù)的分值位置* argc - scoreidx計(jì)算所有的分值，數(shù)據(jù)個數(shù) */elements = c->argc-scoreidx;/* 由于分值和數(shù)據(jù)是成對出現(xiàn)的，這里判斷輸入的個數(shù)是否合法 */if (elements % 2) {addReply(c,shared.syntaxerr);return;}/* 除以2計(jì)算不同<分?jǐn)?shù)，數(shù)據(jù)>對的個數(shù) */elements /= 2; /* 核查幾個選項(xiàng) */.../* 為分值分配內(nèi)存 */scores = zmalloc(sizeof(double)*elements);for (j = 0; j < elements; j++) {/* 將字符串類型轉(zhuǎn)成double */if (getDoubleFromObjectOrReply(c,c->argv[scoreidx+j*2],&scores[j],NULL)!= C_OK) goto cleanup;}/* 在數(shù)據(jù)庫中查找是否存在鍵key，返回鍵對應(yīng)的值 */zobj = lookupKeyWrite(c->db,key);if (zobj == NULL) {/* 如果不存在，創(chuàng)建值 */if (xx) goto reply_to_client; /* No key + XX option: nothing to do. *//* 根據(jù)參數(shù)配置選擇底層采用壓縮字典還是跳表 *//* 如果數(shù)據(jù)長度大于規(guī)定值，則采用跳表，否則選擇壓縮列表 */if (server.zset_max_ziplist_entries == 0 ||server.zset_max_ziplist_value < sdslen(c->argv[scoreidx+1]->ptr)){/* 創(chuàng)建跳表編碼的有序集合 */zobj = createZsetObject();} else {/* 創(chuàng)建壓縮列表編碼的有序集合 */zobj = createZsetZiplistObject();}/* 將鍵值對添加到數(shù)據(jù)庫中，這里值是空的 */dbAdd(c->db,key,zobj);} else {/* 存在鍵key，判斷原先的值是否是用有序集合存儲的 */if (zobj->type != OBJ_ZSET) {addReply(c,shared.wrongtypeerr);goto cleanup;}}/* 對于每個<分?jǐn)?shù)，數(shù)據(jù)>對，將其添加到zobj中 */for (j = 0; j < elements; j++) {/* 第j個分值 */score = scores[j];/* 采用壓縮列表的api執(zhí)行添加操作 */if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {...} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {/* 采用跳表的api添加數(shù)據(jù) */zset *zs = zobj->ptr;zskiplistNode *znode;dictEntry *de;/* 嘗試將數(shù)據(jù)轉(zhuǎn)換成合適的編碼以節(jié)省內(nèi)存 */ele = c->argv[scoreidx+1+j*2] =tryObjectEncoding(c->argv[scoreidx+1+j*2]);/* 采用跳表實(shí)現(xiàn)的有序集合中保存了一個字典，鍵是數(shù)據(jù)，值是分值 */de = dictFind(zs->dict,ele);/* 有序集合中存在要添加的數(shù)據(jù) */if (de != NULL) {if (nx) continue;/* 獲取鍵節(jié)點(diǎn)的數(shù)據(jù)和分值 */curobj = dictGetKey(de);curscore = *(double*)dictGetVal(de);/* incr選項(xiàng)是如果存在該數(shù)據(jù)，則和它對應(yīng)的分值相加 */if (incr) {score += curscore;...}/* 更新數(shù)據(jù)，分值 */if (score != curscore) {/* 先刪除，再添加 */serverAssertWithInfo(c,curobj,zslDelete(zs->zsl,curscore,curobj));/* 跳表插入操作 */znode = zslInsert(zs->zsl,score,curobj);/* 增加數(shù)據(jù)的引用計(jì)數(shù) */incrRefCount(curobj); /* 更新字典中的分值 */dictGetVal(de) = &znode->score; server.dirty++;updated++;}processed++;} else if (!xx) {/* 不存在要添加的數(shù)據(jù)，直接插入 */znode = zslInsert(zs->zsl,score,ele);incrRefCount(ele); /* Inserted in skiplist. */serverAssertWithInfo(c,NULL,dictAdd(zs->dict,ele,&znode->score) == DICT_OK);incrRefCount(ele); /* Added to dictionary. */server.dirty++;added++;processed++;}} else {serverPanic("Unknown sorted set encoding");}}... }

為了不讓函數(shù)太長，這里刪除了一些關(guān)于命令選項(xiàng)的判斷和執(zhí)行，不過還是很長…

獲取排名

在跳表中，看到跳表可以快速計(jì)算<分值，數(shù)據(jù)>的排名。排名是由ZRANK命令完成的，底層由zrankGnericCommand函數(shù)實(shí)現(xiàn)

//t_zset.c /* 返回某個鍵下的指定數(shù)據(jù)的排名 */ void zrankGenericCommand(client *c, int reverse) {/* 第一個參數(shù)是鍵 */robj *key = c->argv[1];/* 第二個參數(shù)是值 */robj *ele = c->argv[2];robj *zobj;unsigned long llen;unsigned long rank;/* 在數(shù)據(jù)庫中查找鍵key是否存在，如果存在，再判斷值是否是由有序集合存儲的 */if ((zobj = lookupKeyReadOrReply(c,key,shared.nullbulk)) == NULL ||checkType(c,zobj,OBJ_ZSET)) return;/* 獲取有序集合中數(shù)據(jù)個數(shù) */llen = zsetLength(zobj);serverAssertWithInfo(c,ele,sdsEncodedObject(ele));/* 根據(jù)底層實(shí)現(xiàn)不同選擇不同的接口 */if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {...} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {/* 如果底層采用跳表實(shí)現(xiàn)，則調(diào)用跳表接口 *//* 獲取鍵key對應(yīng)的有序集合 */zset *zs = zobj->ptr;zskiplist *zsl = zs->zsl;dictEntry *de;double score;ele = c->argv[2];/* 由于跳表通過數(shù)據(jù)查找分值比較慢* 所以Redis采用字典保存<數(shù)據(jù)，分值>對，可通過數(shù)據(jù)快速找到對應(yīng)分值 */de = dictFind(zs->dict,ele);if (de != NULL) {/* 如果有序集合中存在要查找的數(shù)據(jù)，則獲取數(shù)據(jù)的分值 */score = *(double*)dictGetVal(de);/* 調(diào)用跳表接口計(jì)算分值score的排名 */rank = zslGetRank(zsl,score,ele);serverAssertWithInfo(c,ele,rank); /* 根據(jù)選項(xiàng)不同計(jì)算是正向排名還是逆向排名 */if (reverse)addReplyLongLong(c,llen-rank);elseaddReplyLongLong(c,rank-1);} else {addReply(c,shared.nullbulk);}} else {serverPanic("Unknown sorted set encoding");} }

計(jì)算數(shù)據(jù)個數(shù)

zsetLength函數(shù)用于計(jì)算有序集合中數(shù)據(jù)個數(shù)，同樣是調(diào)用跳表或者壓縮列表的接口

//t_zset.c /* 計(jì)算有序集合中數(shù)據(jù)個數(shù) */ unsigned int zsetLength(robj *zobj) {int length = -1;/* 根據(jù)底層實(shí)現(xiàn)不同調(diào)用不同接口 */if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {...} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {/* 跳表中直接保存了數(shù)據(jù)個數(shù) */length = ((zset*)zobj->ptr)->zsl->length;} else {serverPanic("Unknown sorted set encoding");}return length; }

小結(jié)

有序集合中保存的數(shù)據(jù)都是有序的，在對象系統(tǒng)中底層可以由跳表和有序列表實(shí)現(xiàn)，而且跳表實(shí)現(xiàn)還是比較簡單的，而壓縮列表實(shí)現(xiàn)會相對難理解一些

總結(jié)

以上是生活随笔為你收集整理的Redis源码剖析（十二）有序集合跳表实现的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。