redis源码阅读之zset模型

zset模型和之前讲过的模型都不太一样，别的模型底层有多个单独的实现，来回转换，zset这边是由两个实现来存储的，只不过一种是ziplist，另外一种是hashtable+zskiplist的common zset，其中第二种在节省空间上面花了心思——共享底层的redis object，其存储结构如下所示

点击(此处)折叠或打开

1. typedef struct zset {
   
2.     dict *dict;
   
3.     zskiplist *zsl;
   
4. } zset;

在数据插入到hashtable的时候，mapping关系是redis object-->score，与此同时，元素插入到skiplist的时候，mapping关系是score--->redis object， 在后一种情况下，元素自然而然的就按照分数排好序了~~~
由于dict当中只有在注册了keyDestructor的时候才会释放空间，zset为了释放元素的时候更加容易管理，这边只是在zslFreeNode()当中释放空间。删除的时候是从dict里rem，后在skiplist中删除。

这里的转换提供了双向转换，从ziplist到common zset和common zset到ziplist的转换都有，前者的转换基本就是没法满足在redis.conf当中设置的两个条件，具体如下：

点击(此处)折叠或打开

1. # Similarly to hashes and lists, sorted sets are also specially encoded in
   
2. # order to save a lot of space. This encoding is only used when the length and
   
3. # elements of a sorted set are below the following limits:
   
4. zset-max-ziplist-entries 128
   
5. zset-max-ziplist-value 64

点击(此处)折叠或打开

1. int zsetAdd(robj *zobj, double score, sds ele, int *flags, double *newscore) {
   
2.     /* Turn options into simple to check vars. */
   
3.     int incr = (*flags & ZADD_INCR) != 0;
   
4.     int nx = (*flags & ZADD_NX) != 0;
   
5.     int xx = (*flags & ZADD_XX) != 0;
   
6.     *flags = 0; /* We'll return our response flags. */
   
7.     double curscore;
   
8. 
   
9. ...
   
10.             zobj->ptr = zzlInsert(zobj->ptr,ele,score);
    
11.             if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries)
    
12.                 zsetConvert(zobj,OBJ_ENCODING_SKIPLIST);
    
13.             if (sdslen(ele) > server.zset_max_ziplist_value)
    
14.                 zsetConvert(zobj,OBJ_ENCODING_SKIPLIST);
    
15.             if (newscore) *newscore = score;
    
16.             *flags |= ZADD_ADDED;
    
17.             return 1;
18. ...
19.     return 0; /* Never reached. */
    
20. }

转换回到ziplist的代码就只有在union操作的时候才用到了~~~

点击(此处)折叠或打开

1. void zsetConvertToZiplistIfNeeded(robj *zobj, size_t maxelelen) {
   
2.     if (zobj->encoding == OBJ_ENCODING_ZIPLIST) return;
   
3.     zset *zset = zobj->ptr;
   
4. 
   
5.     if (zset->zsl->length <= server.zset_max_ziplist_entries &&
   
6.         maxelelen <= server.zset_max_ziplist_value)
   
7.             zsetConvert(zobj,OBJ_ENCODING_ZIPLIST);
   
8. }

点击(此处)折叠或打开

1. void zunionInterGenericCommand(client *c, robj *dstkey, int op) {
   
2. ...
   
3. 
   
4.     /* parse optional extra arguments */
   
5. ...
   
6.     /* sort sets from the smallest to largest, this will improve our
   
7.      * algorithm's performance */
   
8. ...
   
9.         /* Step 1: Create a dictionary of elements -> aggregated-scores
   
10.          * by iterating one sorted set after the other. */
    
11. ...
    
12.         /* Step 2: convert the dictionary into the final sorted set. */
    
13. ...
    
14.     if (dbDelete(c->db,dstkey))
    
15.         touched = 1;
    
16.     if (dstzset->zsl->length) {
    
17.         zsetConvertToZiplistIfNeeded(dstobj,maxelelen);
    
18.         dbAdd(c->db,dstkey,dstobj);
    
19.         addReplyLongLong(c,zsetLength(dstobj));
    
20.         signalModifiedKey(c->db,dstkey);
    
21.         notifyKeyspaceEvent(NOTIFY_ZSET,
    
22.             (op == SET_OP_UNION) ? "zunionstore" : "zinterstore",
    
23.             dstkey,c->db->id);
    
24.         server.dirty++;
    
25.     } else {
    
26.         decrRefCount(dstobj);
    
27.         addReply(c,shared.czero);
    
28.         if (touched) {
    
29.             signalModifiedKey(c->db,dstkey);
    
30.             notifyKeyspaceEvent(NOTIFY_GENERIC,"del",dstkey,c->db->id);
    
31.             server.dirty++;
    
32.         }
    
33.     }
    
34.     zfree(src);
    
35. }

真正执行转换的代码如下：

点击(此处)折叠或打开

1. void zsetConvert(robj *zobj, int encoding) {
   
2.     zset *zs;
   
3.     zskiplistNode *node, *next;
   
4.     sds ele;
   
5.     double score;
   
6. 
   
7.     if (zobj->encoding == encoding) return;
   
8.     if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
   
9.         unsigned char *zl = zobj->ptr;
   
10.         unsigned char *eptr, *sptr;
    
11.         unsigned char *vstr;
    
12.         unsigned int vlen;
    
13.         long long vlong;
    
14. 
    
15.         if (encoding != OBJ_ENCODING_SKIPLIST)
    
16.             serverPanic("Unknown target encoding");
    
17. 
    
18.         zs = zmalloc(sizeof(*zs));
    
19.         zs->dict = dictCreate(&zsetDictType,NULL);
    
20.         zs->zsl = zslCreate();
    
21. 
    
22.         eptr = ziplistIndex(zl,0);
    
23.         serverAssertWithInfo(NULL,zobj,eptr != NULL);
    
24.         sptr = ziplistNext(zl,eptr);
    
25.         serverAssertWithInfo(NULL,zobj,sptr != NULL);
    
26. 
    
27.         while (eptr != NULL) {
    
28.             score = zzlGetScore(sptr);
    
29.             serverAssertWithInfo(NULL,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));
    
30.             if (vstr == NULL)
    
31.                 ele = sdsfromlonglong(vlong);
    
32.             else
    
33.                 ele = sdsnewlen((char*)vstr,vlen);
    
34. 
    
35.             node = zslInsert(zs->zsl,score,ele);
    
36.             serverAssert(dictAdd(zs->dict,ele,&node->score) == DICT_OK);
    
37.             zzlNext(zl,&eptr,&sptr);
    
38.         }
    
39. 
    
40.         zfree(zobj->ptr);
    
41.         zobj->ptr = zs;
    
42.         zobj->encoding = OBJ_ENCODING_SKIPLIST;
    
43.     } else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
    
44.         unsigned char *zl = ziplistNew();
    
45. 
    
46.         if (encoding != OBJ_ENCODING_ZIPLIST)
    
47.             serverPanic("Unknown target encoding");
    
48. 
    
49.         /* Approach similar to zslFree(), since we want to free the skiplist at
    
50.          * the same time as creating the ziplist. */
    
51.         zs = zobj->ptr;
    
52.         dictRelease(zs->dict);
    
53.         node = zs->zsl->header->level[0].forward;
    
54.         zfree(zs->zsl->header);
    
55.         zfree(zs->zsl);
    
56. 
    
57.         while (node) {
    
58.             zl = zzlInsertAt(zl,NULL,node->ele,node->score);
    
59.             next = node->level[0].forward;
    
60.             zslFreeNode(node);
    
61.             node = next;
    
62.         }
    
63. 
    
64.         zfree(zs);
    
65.         zobj->ptr = zl;
    
66.         zobj->encoding = OBJ_ENCODING_ZIPLIST;
    
67.     } else {
    
68.         serverPanic("Unknown sorted set encoding");
    
69.     }
    
70. }

给外界的接口套路都差不多：先找到数据库，在数据库里对指定的存储操作，操作结束后发送事件通知，这里只列出了rem的代码，因为其他的实在是太长了- -|||，见谅见谅：

点击(此处)折叠或打开

1. void zremCommand(client *c) {
   
2.     robj *key = c->argv[1];
   
3.     robj *zobj;
   
4.     int deleted = 0, keyremoved = 0, j;
   
5. 
   
6.     if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
   
7.         checkType(c,zobj,OBJ_ZSET)) return;
   
8. 
   
9.     for (j = 2; j < c->argc; j++) {
   
10.         if (zsetDel(zobj,c->argv[j]->ptr)) deleted++;
    
11.         if (zsetLength(zobj) == 0) {
    
12.             dbDelete(c->db,key);
    
13.             keyremoved = 1;
    
14.             break;
    
15.         }
    
16.     }
    
17. 
    
18.     if (deleted) {
    
19.         notifyKeyspaceEvent(NOTIFY_ZSET,"zrem",key,c->db->id);
    
20.         if (keyremoved)
    
21.             notifyKeyspaceEvent(NOTIFY_GENERIC,"del",key,c->db->id);
    
22.         signalModifiedKey(c->db,key);
    
23.         server.dirty += deleted;
    
24.     }
    
25.     addReplyLongLong(c,deleted);
    
26. }

还请各位多提宝贵意见~~~