[大数据]Redis_ZSet

ZSet简介

ZSet是redis的有序集合实现，包括一个为了字典（按照key直接取值）和一个跳表（按照排名取）

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

typedef struct zskiplist {
    struct zskiplistNode *header, *tail;
    unsigned long length;
    int level;
} zskiplist;

快表节点

跳表的节点，zskiplistNode包含分数、redis键值对象，后退指针，和一个多层数组
多层数组level存放跳表每一层中当前节点的前一个节点forward，以及forward节点与当前节点在该层的跨度span

typedef struct zskiplistNode {
    robj *obj;
    double score;
    struct zskiplistNode *backward;
    struct zskiplistLevel {
        struct zskiplistNode *forward;
        unsigned int span;
    } level[];
} zskiplistNode;

创建快表

zskiplist *zslCreate(void) {
    int j;
    zskiplist *zsl;
    zsl = zmalloc(sizeof(*zsl));
    zsl->level = 1;
    zsl->length = 0;
   //初始化头节点为分数0
    zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
        zsl->header->level[j].forward = NULL;
        zsl->header->level[j].span = 0;
    }
    zsl->header->backward = NULL;
    zsl->tail = NULL;
    return zsl;
}

插入节点

省略部分流程代码

zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;
    x = zsl->header;
    //后面的插入节点处理是随机到某一深度
    //也就是节点可能在高层（比如0）有，但是在深层（比如level-1）没有
    //所以计算位置和连接节点自底向上快一点
    for (i = zsl->level-1; i >= 0; i--) {
    	//rank记录x在每一层统计后的排名，主要是为了后面插入时更新链接节点的span，不重复计算
        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
        
        while (x->level[i].forward &&
            (x->level[i].forward->score < score )) {
            //while中i不会变，是在i这一层横向移动到本层中比新插入分数大的节点
            //跨越该节点增加本层排名
            rank[i] += x->level[i].span;
            x = x->level[i].forward;
        }
        //记录在i这一层比新插入分数大的节点
        update[i] = x;
    }

	//概率随机插入到某一层深度
    level = zslRandomLevel();
    if (level > zsl->level) {
        for (i = zsl->level; i < level; i++) {
            rank[i] = 0;
            update[i] = zsl->header;
            update[i]->level[i].span = zsl->length;
        }
        zsl->level = level;
    }

	//x现在是新节点
    x = zslCreateNode(level,score,obj);

    for (i = 0; i < level; i++) {
    	//把刚刚计算过的，跳跃表每层比x分数大的节点，链接到x，作为x的该层前一个节点
        x->level[i].forward = update[i]->level[i].forward;
        update[i]->level[i].forward = x;
        //更新跨度
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
    }
     //超出的层数不插入，只更新span
    for (i = level; i < zsl->level; i++) {
        update[i]->level[i].span++;
    }
	//更新backward和tail、length
    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;
}

删除节点

void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
    int i;
	//删除相对简单，更新span，移除节点
    for (i = 0; i < zsl->level; i++) {
        if (update[i]->level[i].forward == x) {
            update[i]->level[i].span += x->level[i].span - 1;
            update[i]->level[i].forward = x->level[i].forward;
        } else {
            update[i]->level[i].span -= 1;
        }
    }

    // 更新backward、tail、length、level
    if (x->level[0].forward) {
        x->level[0].forward->backward = x->backward;
    } else {
        zsl->tail = x->backward;
    }
    while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
        zsl->level--;
    zsl->length--;
}

查询节点

zskiplistNode* zslGetElementByRank(zskiplist *zsl, unsigned long rank) {
    zskiplistNode *x;
    unsigned long traversed = 0;
    int i;
    x = zsl->header;
    //先从最底层找（元素少），跳到没有排名可以跳了，再到上层
    for (i = zsl->level-1; i >= 0; i--) {
        while (x->level[i].forward && (traversed + x->level[i].span) <= rank)
        {
            traversed += x->level[i].span;
            x = x->level[i].forward;
        }
        if (traversed == rank) {
            return x;
        }

    }
    return NULL;
}

更多文章，请搜索公众号歪歪梯Club