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

本篇主要涉及的是有序集合添加數據的命令，后面會看到，在命令的底層實現中，實際上還是調用跳表的接口

存儲結構

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

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

有序集合操作

添加數據

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

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

為了不讓函數太長，這里刪除了一些關于命令選項的判斷和執行，不過還是很長…

獲取排名

在跳表中，看到跳表可以快速計算<分值，數據>的排名。排名是由ZRANK命令完成的，底層由zrankGnericCommand函數實現

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

計算數據個數

zsetLength函數用于計算有序集合中數據個數，同樣是調用跳表或者壓縮列表的接口

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

小結

有序集合中保存的數據都是有序的，在對象系統中底層可以由跳表和有序列表實現，而且跳表實現還是比較簡單的，而壓縮列表實現會相對難理解一些

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。