Redis源码剖析有序集合对象t_zset实现

比眉伴天荒 2022-02-26 00:50 199阅读 0赞

有序集合对象其实跟集合对象类似，只不过它多了一个score的参数，集合中的每个元素都有一个分值，在集合中元素是按照score排序的。

有序集合的底层编码也是有两种实现，压缩列表REDIS\_ENCODING\_ZIPLIST以及跳跃表REDIS\_ENCODING\_SKIPLIST。和集合的一样，有序集合的编码方式是通过检查第一个被加入的元素来决定的。

# ZSET结构 #

/* RedisObject结构 */
    typedef struct redisObject {
        unsigned type:4;  // OBJ_ZSET表示有序集合对象
        unsigned encoding:4;  // 编码字段为OBJ_ENCODING_ZIPLIST或OBJ_ENCODING_SKIPLIST
        unsigned lru:LRU_BITS; // LRU_BITS为24位
        int refcount;
        void *ptr;  // 指向数据部分
    } robj;

同其他的对象一样， zset结构也是存储在redisObject结构体中，通过指定 type= OBJ\_ZSET 来确定这是一个有序集合对象，当是一个有序集合对象的时候，配套的endoding只能有对应的两种取值。

# ZIPLIST编码的有序集合 #

当用REDIS\_ENCODING\_ZIPLIST作为有序集合的底层编码时，有序集合中的元素按score值从小到大排序，先保存value，在保存score，示意图如下

# SKIPLIST编码的有序集合 #

typedef struct zset {
        dict *dict;
        zskiplist *zsl;
    } zset;

当有序集合用REDIS\_ENCODING\_SKIPLIST编码的时候，并不仅仅是用了跳跃表skiplist，还用到了字典dict。为什么要这么做呢。 跳跃表的优点是能够在O(log\_N\_)的期望时间内根据分值score对元素进行定位，对于一些范围查找命令，比如ZRANGE能够较好的支持。

但是如果要取出对应元素的分值，或者查看没有某个元素是否在有序集合内，单纯靠跳跃表就不够用了， 因为跳跃表是根据score组织的，不是根据元素值组织的。所以在有序集合中另外用了一个dict来支持这些操作。dict中key就是元素的值，value就是score。这样就能够在O(1)的复杂度内找到对应的分值，或者判断一个元素是否在有序集合中。

# ZSET编码转换 #

如果一个有序集合一开始是用压缩列表REDIS\_ENCODING\_ZIPLIST作为底层编码，只要满足下边的条件，就会将底层编码转换为REDIS\_ENCODING\_SKIPLIST：

*  ziplist 所保存的元素数量超过服务器属性 server.zset\_max\_ziplist\_entries 的值（默认值为 128 ）
 *  新添加元素的长度大于服务器属性 server.zset\_max\_ziplist\_value 的值（默认值为 64 ）

void zsetConvert(robj *zobj, int encoding) {
        zset *zs;
        zskiplistNode *node, *next;
        sds ele;
        double score;
        // 如果已经是目标编码格式，返回
        if (zobj->encoding == encoding) return;
        // 编码格式从ZIPLIST转成SKIPLIST
        if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
            unsigned char *zl = zobj->ptr;
            unsigned char *eptr, *sptr;
            unsigned char *vstr;
            unsigned int vlen;
            long long vlong;
            // 如果目标编码格式不对，返回错误
            if (encoding != OBJ_ENCODING_SKIPLIST)
                serverPanic("Unknown target encoding");
            // 创建新的zset，底层为SKIPLIST编码，所以需要两个结构体：一个dict和一个skiplist
            zs = zmalloc(sizeof(*zs));
            zs->dict = dictCreate(&zsetDictType,NULL);
            zs->zsl = zslCreate();
    
            eptr = ziplistIndex(zl,0);
            serverAssertWithInfo(NULL,zobj,eptr != NULL);
            sptr = ziplistNext(zl,eptr);
            serverAssertWithInfo(NULL,zobj,sptr != NULL);
            // 循环遍历ZIPLIST
            while (eptr != NULL) {
                // 获取score
                score = zzlGetScore(sptr);
                serverAssertWithInfo(NULL,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));
                // 获得对应的元素的值
                if (vstr == NULL)
                    ele = sdsfromlonglong(vlong);
                else
                    ele = sdsnewlen((char*)vstr,vlen);
                // 根据元素值和score新建node，并插入到dict中
                node = zslInsert(zs->zsl,score,ele);
                serverAssert(dictAdd(zs->dict,ele,&node->score) == DICT_OK);
                zzlNext(zl,&eptr,&sptr);
            }
            // 将zobj指向新的zset
            zfree(zobj->ptr);
            zobj->ptr = zs;
            zobj->encoding = OBJ_ENCODING_SKIPLIST;
        } else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
            // 编码格式从SKIPLIST转换成ZIPLIST
            unsigned char *zl = ziplistNew();
            // 检查目标编码格式，错误退出
            if (encoding != OBJ_ENCODING_ZIPLIST)
                serverPanic("Unknown target encoding");
    
            /* Approach similar to zslFree(), since we want to free the skiplist at
             * the same time as creating the ziplist. */
            zs = zobj->ptr;
            // 释放dict的空间
            dictRelease(zs->dict);
            // skiplist的头节点空间
            node = zs->zsl->header->level[0].forward;
            // 释放表头
            zfree(zs->zsl->header);
            zfree(zs->zsl);
            // 遍历跳跃表
            while (node) {
                // 将元素ele和score添加到ziplist中
                zl = zzlInsertAt(zl,NULL,node->ele,node->score);
                next = node->level[0].forward;
                zslFreeNode(node);
                node = next;
            }
            // 释放zs，并集鞥zobj指向新的zl
            zfree(zs);
            zobj->ptr = zl;
            zobj->encoding = OBJ_ENCODING_ZIPLIST;
        } else {
            serverPanic("Unknown sorted set encoding");
        }
    }

# ZSET命令 #

<table> 
 <thead> 
  <tr> 
   <th>命令</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ZADD key score member [[score member] [score member] …]</td> 
   <td>将一个或多个 member 元素及其 score 值加入到有序集 key 当中</td> 
  </tr> 
  <tr> 
   <td>zcard</td> 
   <td>返回有序集 key 的基数</td> 
  </tr> 
  <tr> 
   <td>ZCOUNT key min max</td> 
   <td>返回有序集 key 中， score 值在 min 和 max 之间(默认包括 score 值等于 min 或 max )的成员的数量</td> 
  </tr> 
  <tr> 
   <td>ZINCRBY key increment member</td> 
   <td>为有序集 key 的成员 member 的 score 值加上增量 increment</td> 
  </tr> 
  <tr> 
   <td>zrange</td> 
   <td>返回有序集 key 中，指定区间内的成员</td> 
  </tr> 
  <tr> 
   <td>zrevrange</td> 
   <td>返回有序集 key 中，指定区间内的成员</td> 
  </tr> 
  <tr> 
   <td>zrangeByScore</td> 
   <td>返回有序集 key 中，所有 score 值介于 min 和 max 之间(包括等于 min 或 max )的成员</td> 
  </tr> 
  <tr> 
   <td>zrank</td> 
   <td>返回有序集 key 中成员 member 的排名。其中有序集成员按 score 值递增(从小到大)顺序排列</td> 
  </tr> 
  <tr> 
   <td>zrevrank</td> 
   <td>返回有序集 key 中成员 member 的排名。其中有序集成员按 score 值递减(从大到小)排序</td> 
  </tr> 
  <tr> 
   <td>zrem</td> 
   <td>移除有序集 key 中的一个或多个成员，不存在的成员将被忽略</td> 
  </tr> 
  <tr> 
   <td>zscore</td> 
   <td>返回有序集 key 中，成员 member 的 score 值</td> 
  </tr> 
 </tbody> 
</table>

# ZSET命令实现 #

## ZADD接口实现 ##

区别于其他有多种底层编码格式的实现（比如集合SET），有序集合不是在一个函数中区别不同的底层编码来实现功能，而是分别搞了两套机制，比如ZADD命令，有一个压缩列表编码的 zzlInsert 函数以及跳跃表编码的 zslInsert 函数。在命令一进来的时候，就根据不同的编码格式，调用不同的函数实现。

ZIPLIST的中插入接口

/* Insert (element,score) pair in ziplist. This function assumes the element is
     * not yet present in the list. */
    unsigned char *zzlInsert(unsigned char *zl, sds ele, double score) {
        unsigned char *eptr = ziplistIndex(zl,0), *sptr;
        double s;
        // 循环遍历ZIPLIST
        while (eptr != NULL) {
            sptr = ziplistNext(zl,eptr);
            serverAssert(sptr != NULL);
            // 获取分值
            s = zzlGetScore(sptr);
            // 如果分值大于score，说明已经找到了要插入的位置
            if (s > score) {
                /* First element with score larger than score for element to be
                 * inserted. This means we should take its spot in the list to
                 * maintain ordering. */
                // 在对应位置插入元素和score
                zl = zzlInsertAt(zl,eptr,ele,score);
                break;
            } else if (s == score) {
                /* Ensure lexicographical ordering for elements. */
                // 如果分值相同，按字典序排列
                if (zzlCompareElements(eptr,(unsigned char*)ele,sdslen(ele)) > 0) {
                    zl = zzlInsertAt(zl,eptr,ele,score);
                    break;
                }
            }
    
            /* Move to next element. */
            eptr = ziplistNext(zl,sptr);
        }
    
        /* Push on tail of list when it was not yet inserted. */
        // 如果到了最后，说明前边的score都比目标要小，直接在尾部插入
        if (eptr == NULL)
            zl = zzlInsertAt(zl,NULL,ele,score);
        return zl;
    }

SKIPLIST的插入接口在跳跃表那节介绍过，重新贴一下

/* Insert a new node in the skiplist. Assumes the element does not already
     * exist (up to the caller to enforce that). The skiplist takes ownership
     * of the passed SDS string 'ele'. */
     // 跳跃表插入元素
    zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
        zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
        unsigned int rank[ZSKIPLIST_MAXLEVEL];
        int i, level;
    
        serverAssert(!isnan(score));   　// 判断是否为数字
        x = zsl->header;
        // 从最高的level， 也即跨度最大的level开始查找结点
        for (i = zsl->level-1; i >= 0; i--) {
            /* store rank that is crossed to reach the insert position */
            // 当前是否是最高层， 如果是最高层，rank[i]=0，否则，复制上一层的数值
            rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
            // 如果当前结点的score值小于传入的score 或者 当前score相等，但是结点的对象不相等
            while (x->level[i].forward &&
                    (x->level[i].forward->score < score ||
                        (x->level[i].forward->score == score &&
                        sdscmp(x->level[i].forward->ele,ele) < 0)))
            {
                // 将当前一层的跨度加到rank[i]
                rank[i] += x->level[i].span;
                // 在当前层中向前查找
                x = x->level[i].forward;
            }
            // 当前层位于插入位置前的结点x放入update数组
            update[i] = x;
        }
        /* we assume the element is not already inside, since we allow duplicated
         * scores, reinserting the same element should never happen since the
         * caller of zslInsert() should test in the hash table if the element is
         * already inside or not. */
        // 随机生成小于32的层数
        level = zslRandomLevel();
        // 如果生成的层数大于当前的层数
        if (level > zsl->level) {
            for (i = zsl->level; i < level; i++) {
                // 设定rank数组中大于原level层以上的值为0
                // 同时设定update数组大于原level层以上的数据
                rank[i] = 0;
                update[i] = zsl->header;
                update[i]->level[i].span = zsl->length;
            }
            zsl->level = level;
        }
        // 创建层数为level的新结点
        x = zslCreateNode(level,score,ele);
        for (i = 0; i < level; i++) {
            // 将每一层的前置结点的后续结点指向新结点， 同时设置新结点的后续结点
            x->level[i].forward = update[i]->level[i].forward;
            update[i]->level[i].forward = x;
    
            /* update span covered by update[i] as x is inserted here */
            // 更新每一层的前置结点和新结点的跨度
            x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
            update[i]->level[i].span = (rank[0] - rank[i]) + 1;
        }
    
        /* increment span for untouched levels */
        for (i = level; i < zsl->level; i++) {
            update[i]->level[i].span++;
        }
    
        // 根据最低层的前序结点是否是header结点来设置当前新结点的向后指针
        x->backward = (update[0] == zsl->header) ? NULL : update[0];
        if (x->level[0].forward)
            x->level[0].forward->backward = x;
        else
            zsl->tail = x;
        zsl->length++;
        return x;
    }