顺风车运营研发团队 张仕华

ZADD
ZADD key [NX|XX] [CH] [INCR]score member [score member ...]

将元素及对应分值添加到一个有序集合中

NX:不更新已经存在的key,只增加新元素

XX:只更新已经存在的key,不增加新元素

CH:abbr:changed.不指定时只返回新增的元素个数,指定时返回新增的和更新的元素个数之和

INCR:参考zincrby

//通过第二个参数区分是zadd还是zincrby
void zaddCommand(client *c) {zaddGenericCommand(c,ZADD_NONE);
}

/* This generic command implements both ZADD and ZINCRBY. */
//zadd和zincrby两个命令都是调用这个函数
void zaddGenericCommand(client *c, int flags) {static char *nanerr = "resulting score is not a number (NaN)";robj *key = c->argv[1];robj *zobj;sds ele;double score = 0, *scores = NULL;int j, elements;int scoreidx = 0;/* The following vars are used in order to track what the command actually* did during the execution, to reply to the client and to trigger the* notification of keyspace change. */int added = 0;      /* Number of new elements added. */int updated = 0;    /* Number of elements with updated score. */int processed = 0;  /* Number of elements processed, may remain zero withoptions like XX. *//* Parse options. At the end 'scoreidx' is set to the argument position* of the score of the first score-element pair. */scoreidx = 2;//从第二个参数开始处理.先处理nx,xx,ch,incr参数while(scoreidx < c->argc) {char *opt = c->argv[scoreidx]->ptr;if (!strcasecmp(opt,"nx")) flags |= ZADD_NX;else if (!strcasecmp(opt,"xx")) flags |= ZADD_XX;else if (!strcasecmp(opt,"ch")) flags |= ZADD_CH;else if (!strcasecmp(opt,"incr")) flags |= ZADD_INCR;else break;scoreidx++;}/* Turn options into simple to check vars. *///从flag中取出相应的标志赋给独立的变量int incr = (flags & ZADD_INCR) != 0;int nx = (flags & ZADD_NX) != 0;int xx = (flags & ZADD_XX) != 0;int ch = (flags & ZADD_CH) != 0;/* After the options, we expect to have an even number of args, since* we expect any number of score-element pairs. *///member和score是一一对应的,所以肯定是2的倍数.所以如果不是2的倍数或者根本//没有member和score,直接返回命令语法错误elements = c->argc-scoreidx;if (elements % 2 || !elements) {addReply(c,shared.syntaxerr);return;}//elements赋值为有多少对<element,score>elements /= 2; /* Now this holds the number of score-element pairs. *//* Check for incompatible options. *///nx和xxflag互斥,二者不能同时出现if (nx && xx) {addReplyError(c,"XX and NX options at the same time are not compatible");return;}//若有incr标志,则只能有一对<element,score>//为什么不能是多对?if (incr && elements > 1) {addReplyError(c,"INCR option supports a single increment-element pair");return;}/* Start parsing all the scores, we need to emit any syntax error* before executing additions to the sorted set, as the command should* either execute fully or nothing at all. *///依次检查每一个分数值scores = zmalloc(sizeof(double)*elements);for (j = 0; j < elements; j++) {//该函数中会检查score是否是合法的double类型的值if (getDoubleFromObjectOrReply(c,c->argv[scoreidx+j*2],&scores[j],NULL)!= C_OK) goto cleanup;}/* Lookup the key and create the sorted set if does not exist. *///根据key查找对应的有序集合的valuezobj = lookupKeyWrite(c->db,key);//key不存在if (zobj == NULL) {//如果设置了xx这个flag,直接返回错误if (xx) goto reply_to_client; /* No key + XX option: nothing to do. *///根据redis的配置,如果有序集合设置了不使用ziplist存储或者说第一个插入元素的长度大于//设置的最大ziplist的元素长度值,则使用跳跃表存储否则使用ziplistif (server.zset_max_ziplist_entries == 0 ||server.zset_max_ziplist_value < sdslen(c->argv[scoreidx+1]->ptr)){zobj = createZsetObject();} else {zobj = createZsetZiplistObject();}//把key,zobj插入字典dbAdd(c->db,key,zobj);//key存在} else {//如果不是有序集合,直接返回错误if (zobj->type != OBJ_ZSET) {addReply(c,shared.wrongtypeerr);goto cleanup;}}//elements是<member,score>对数for (j = 0; j < elements; j++) {double newscore;score = scores[j];//retflags设置为前文中的flags变量int retflags = flags;ele = c->argv[scoreidx+1+j*2]->ptr;//每次遍历,score是分数,ele是member.调用zsetadd插入zobjint retval = zsetAdd(zobj, score, ele, &retflags, &newscore);if (retval == 0) {addReplyError(c,nanerr);goto cleanup;}//根据retflags,即一个元素是更新还是新加入,还是未做处理(即member存在,并且//score值与新设置的一致),更新相应的计数变量(这些变量最后会返回给客户端)if (retflags & ZADD_ADDED) added++;if (retflags & ZADD_UPDATED) updated++;if (!(retflags & ZADD_NOP)) processed++;score = newscore;}server.dirty += (added+updated);
//通过命令中的flag,返回给客户端不同的值
reply_to_client:if (incr) { /* ZINCRBY or INCR option. */if (processed)addReplyDouble(c,score);elseaddReply(c,shared.nullbulk);} else { /* ZADD. */addReplyLongLong(c,ch ? added+updated : added);}//如果有更新或者新加,需要执行相应的watch key的通知及keyspace的通知
cleanup:zfree(scores);if (added || updated) {signalModifiedKey(c->db,key);notifyKeyspaceEvent(NOTIFY_ZSET,incr ? "zincr" : "zadd", key, c->db->id);}
}

ZINCRBY
ZINCRBY key increment member

如果key存在,就给相应member的score增加increment

否则直接给key设置分数为increment

//与zadd调用同一个函数,相当于zadd key incr,把incr flag置位
void zincrbyCommand(client *c) {zaddGenericCommand(c,ZADD_INCR);
}

ZCARD
ZCARD key

返回有序集合的元素个数

void zcardCommand(client *c) {robj *key = c->argv[1];robj *zobj;//查找key对应的valueif ((zobj = lookupKeyReadOrReply(c,key,shared.czero)) == NULL ||checkType(c,zobj,OBJ_ZSET)) return;//通过zsetLength获取zobj中的元素个数addReplyLongLong(c,zsetLength(zobj));
}

unsigned int zsetLength(const robj *zobj) {int length = -1;//如果是ziplist,通过zzlLength函数获取长度//如果长度字段中的值小于UINT16_MAX，直接返回长度。否则需要遍历获取长度if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {length = zzlLength(zobj->ptr);//如果是skiplist,直接返回zsl->length} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {length = ((const zset*)zobj->ptr)->zsl->length;} else {serverPanic("Unknown sorted set encoding");}return length;
}

ZCOUNT
ZCOUNT key min max

返回key中score值在min和max之间的元素个数

其中min和max可以加(,如 zcount key (5 (10

加左括号表示不包含。不加表示包含

void zcountCommand(client *c) {robj *key = c->argv[1];robj *zobj;zrangespec range;int count = 0;/* Parse the range arguments *///判定范围.并将最大最小及是否包含写入range结构体中if (zslParseRange(c->argv[2],c->argv[3],&range) != C_OK) {addReplyError(c,"min or max is not a float");return;}/* Lookup the sorted set */if ((zobj = lookupKeyReadOrReply(c, key, shared.czero)) == NULL ||checkType(c, zobj, OBJ_ZSET)) return;//判断zobj底层编码是ziplist还是skiplistif (zobj->encoding == OBJ_ENCODING_ZIPLIST) {unsigned char *zl = zobj->ptr;unsigned char *eptr, *sptr;double score;//找出第一个在范围之内的元素/* Use the first element in range as the starting point */eptr = zzlFirstInRange(zl,&range);/* No "first" element */if (eptr == NULL) {addReply(c, shared.czero);return;}/* First element is in range *///ziplist中member和score是两个entry,并且member之后保存着score//整体顺序是按score从小到大排列,score相同时,按member的字典序排列sptr = ziplistNext(zl,eptr);//所以此处从第一个元素的下一个entry处获取scorescore = zzlGetScore(sptr);serverAssertWithInfo(c,zobj,zslValueLteMax(score,&range));/* Iterate over elements in range *///迭代这个ziplist,如果score满足要求,则count++并且继续迭代,否则跳出//最后会返回countwhile (eptr) {score = zzlGetScore(sptr);/* Abort when the node is no longer in range. */if (!zslValueLteMax(score,&range)) {break;} else {count++;zzlNext(zl,&eptr,&sptr);}}} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {zset *zs = zobj->ptr;zskiplist *zsl = zs->zsl;zskiplistNode *zn;unsigned long rank;//如果是跳表,也是先取出第一个元素/* Find first element in range */zn = zslFirstInRange(zsl, &range);/* Use rank of first element, if any, to determine preliminary count */if (zn != NULL) {//获取第一个元素的排名rank = zslGetRank(zsl, zn->score, zn->ele);count = (zsl->length - (rank - 1));//如果最大值大于zsl中的最大值,则此count就是要找的个数/* Find last element in range */zn = zslLastInRange(zsl, &range);/* Use rank of last element, if any, to determine the actual count */if (zn != NULL) {//如果最大值小于zsl中的最大值，则首先找到最后一个元素的rankrank = zslGetRank(zsl, zn->score, zn->ele);//重新计算count,与之前的计算公式合并之后为//count = (zsl->length-(rankmin-1))-(zsl->length-rankmax))//      = rankmax-rankmin+1count -= (zsl->length - rank);}}} else {serverPanic("Unknown sorted set encoding");}//返回countaddReplyLongLong(c, count);
}

ZRANGEBYSCORE
ZRANGEBYSCORE key min max WITHSCORES

获取有序结合中分值位于 min和max之间的所有元素

withscores:将member 和 score一起返回

limit offset count:从偏移offset开始获取count个元素

min和max可以为 -inf,+inf,分别表示负无穷和正无穷

//入口函数
void zrangebyscoreCommand(client *c) {genericZrangebyscoreCommand(c,0);
}

/* This command implements ZRANGEBYSCORE, ZREVRANGEBYSCORE. */
void genericZrangebyscoreCommand(client *c, int reverse) {zrangespec range;robj *key = c->argv[1];robj *zobj;long offset = 0, limit = -1;int withscores = 0;unsigned long rangelen = 0;void *replylen = NULL;int minidx, maxidx;//该函数同时用于zrangbyscore和zrevrangebyscore//二者通过函数中的reverse参数标识//正序时第二个参数是min，第三个参数是max,逆序反之/* Parse the range arguments. */if (reverse) {/* Range is given as [max,min] */maxidx = 2; minidx = 3;} else {/* Range is given as [min,max] */minidx = 2; maxidx = 3;}//将参数解析出来赋值到range变量if (zslParseRange(c->argv[minidx],c->argv[maxidx],&range) != C_OK) {addReplyError(c,"min or max is not a float");return;}/* Parse optional extra arguments. Note that ZCOUNT will exactly have* 4 arguments, so we'll never enter the following code path. */if (c->argc > 4) {int remaining = c->argc - 4;int pos = 4;//解析withscores和limit参数while (remaining) {if (remaining >= 1 && !strcasecmp(c->argv[pos]->ptr,"withscores")) {pos++; remaining--;withscores = 1;} else if (remaining >= 3 && !strcasecmp(c->argv[pos]->ptr,"limit")) {if ((getLongFromObjectOrReply(c, c->argv[pos+1], &offset, NULL)!= C_OK) ||(getLongFromObjectOrReply(c, c->argv[pos+2], &limit, NULL)!= C_OK)){return;}pos += 3; remaining -= 3;} else {addReply(c,shared.syntaxerr);return;}}}/* Ok, lookup the key and get the range */if ((zobj = lookupKeyReadOrReply(c,key,shared.emptymultibulk)) == NULL ||checkType(c,zobj,OBJ_ZSET)) return;//按zset底层编码是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;double score;/* If reversed, get the last node in range as starting point. *///按正序还是逆序分别取最后一个值或者第一个值if (reverse) {eptr = zzlLastInRange(zl,&range);} else {eptr = zzlFirstInRange(zl,&range);}/* No "first" element in the specified interval. */if (eptr == NULL) {addReply(c, shared.emptymultibulk);return;}/* Get score pointer for the first element. */serverAssertWithInfo(c,zobj,eptr != NULL);sptr = ziplistNext(zl,eptr);/* We don't know in advance how many matching elements there are in the* list, so we push this object that will represent the multi-bulk* length in the output buffer, and will "fix" it later */replylen = addDeferredMultiBulkLength(c);/* If there is an offset, just traverse the number of elements without* checking the score because that is done in the next loop. *///如果有offset,先偏移相应的元素//注意此处zzlNext传入了两个指针,会一次偏移一个<member,score>对//注意此处offset初始值是0,如果没指定则不会进入此处循环while (eptr && offset--) {if (reverse) {zzlPrev(zl,&eptr,&sptr);} else {zzlNext(zl,&eptr,&sptr);}}//如果有limit,则进入循环.取limit次.limit的初始值为-1,即使没指定,也会进入循环//直到eptr为null或者循环中break掉while (eptr && limit--) {score = zzlGetScore(sptr);/* Abort when the node is no longer in range. *///不在范围之内时break掉if (reverse) {if (!zslValueGteMin(score,&range)) break;} else {if (!zslValueLteMax(score,&range)) break;}/* We know the element exists, so ziplistGet should always succeed */serverAssertWithInfo(c,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));//取出相应的值.可能为str,赋值给vstr,长度为vlen,或者为整型,赋值给vlongrangelen++;if (vstr == NULL) {addReplyBulkLongLong(c,vlong);} else {addReplyBulkCBuffer(c,vstr,vlen);}//如果设置了withscores标志,则返回分数if (withscores) {addReplyDouble(c,score);}//开始迭代下一个节点/* Move to next node */if (reverse) {zzlPrev(zl,&eptr,&sptr);} else {zzlNext(zl,&eptr,&sptr);}}} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {zset *zs = zobj->ptr;zskiplist *zsl = zs->zsl;zskiplistNode *ln;//同ziplist,先查找起始或最终节点/* If reversed, get the last node in range as starting point. */if (reverse) {ln = zslLastInRange(zsl,&range);} else {ln = zslFirstInRange(zsl,&range);}/* No "first" element in the specified interval. */if (ln == NULL) {addReply(c, shared.emptymultibulk);return;}/* We don't know in advance how many matching elements there are in the* list, so we push this object that will represent the multi-bulk* length in the output buffer, and will "fix" it later *///返回客户端时先返回元素个数,但此处并不知道需要返回多少个元素,所以先占个位置//replylen是存储len字段的指针replylen = addDeferredMultiBulkLength(c);/* If there is an offset, just traverse the number of elements without* checking the score because that is done in the next loop. *///处理offset.向前或向后skipwhile (ln && offset--) {if (reverse) {ln = ln->backward;} else {ln = ln->level[0].forward;}}//处理limitwhile (ln && limit--) {/* Abort when the node is no longer in range. */if (reverse) {if (!zslValueGteMin(ln->score,&range)) break;} else {if (!zslValueLteMax(ln->score,&range)) break;}rangelen++;addReplyBulkCBuffer(c,ln->ele,sdslen(ln->ele));if (withscores) {addReplyDouble(c,ln->score);}/* Move to next node */if (reverse) {ln = ln->backward;} else {ln = ln->level[0].forward;}}} else {serverPanic("Unknown sorted set encoding");}//如果有withscores参数,返回给客户端的字符串数量是2倍if (withscores) {rangelen *= 2;}//将rangelen放入replylen指向的位置,返回给客户端setDeferredMultiBulkLength(c, replylen, rangelen);
}

ZRANK
ZRANK key member

返回有序集合中元素member的rank

以0为起始rank,元素分数从低到高

zrevrank,元素分数从高到低

void zrankGenericCommand(client *c, int reverse) {robj *key = c->argv[1];robj *ele = c->argv[2];robj *zobj;long rank;//通过key找出有序集合的value zobjif ((zobj = lookupKeyReadOrReply(c,key,shared.nullbulk)) == NULL ||checkType(c,zobj,OBJ_ZSET)) return;serverAssertWithInfo(c,ele,sdsEncodedObject(ele));//在zobj中查找ele(第二个参数member)rank = zsetRank(zobj,ele->ptr,reverse);if (rank >= 0) {addReplyLongLong(c,rank);} else {addReply(c,shared.nullbulk);}
}void zrankCommand(client *c) {zrankGenericCommand(c, 0);
}

long zsetRank(robj *zobj, sds ele, int reverse) {unsigned long llen;unsigned long rank;llen = zsetLength(zobj);//ziplist从前往后遍历,比较entry中的元素与ele,每次将rank++if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {unsigned char *zl = zobj->ptr;unsigned char *eptr, *sptr;eptr = ziplistIndex(zl,0);serverAssert(eptr != NULL);sptr = ziplistNext(zl,eptr);serverAssert(sptr != NULL);rank = 1;while(eptr != NULL) {if (ziplistCompare(eptr,(unsigned char*)ele,sdslen(ele)))break;rank++;zzlNext(zl,&eptr,&sptr);}if (eptr != NULL) {//如果是逆序取,直接将llen-rank就是逆向的rankif (reverse)return llen-rank;elsereturn rank-1;} else {return -1;}//skiplist通过zslGetRank获取rank,具体过程为跳表查找,将相应路过节点的span相加} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {zset *zs = zobj->ptr;zskiplist *zsl = zs->zsl;dictEntry *de;double score;de = dictFind(zs->dict,ele);if (de != NULL) {score = *(double*)dictGetVal(de);rank = zslGetRank(zsl,score,ele);/* Existing elements always have a rank. */serverAssert(rank != 0);//逆向取rankif (reverse)return llen-rank;elsereturn rank-1;} else {return -1;}} else {serverPanic("Unknown sorted set encoding");}
}

ZREM
ZREM key member [member ...]

从有序集合中删除相应的member

void zremCommand(client *c) {robj *key = c->argv[1];robj *zobj;int deleted = 0, keyremoved = 0, j;//根据key找到对应的zobjif ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||checkType(c,zobj,OBJ_ZSET)) return;//依次删除相应的元素,每次删除之后检查zset是否为空,如果为空,删掉该key,并且breakfor (j = 2; j < c->argc; j++) {if (zsetDel(zobj,c->argv[j]->ptr)) deleted++;if (zsetLength(zobj) == 0) {dbDelete(c->db,key);keyremoved = 1;break;}}//如果确实有member被删除掉,通知keyspace zrem事件//如果zset整个都被删除了,通知keyspace del事件if (deleted) {notifyKeyspaceEvent(NOTIFY_ZSET,"zrem",key,c->db->id);if (keyremoved)notifyKeyspaceEvent(NOTIFY_GENERIC,"del",key,c->db->id);signalModifiedKey(c->db,key);server.dirty += deleted;}//返回给客户端实际删除的member个数addReplyLongLong(c,deleted);
}

/* Delete the element 'ele' from the sorted set, returning 1 if the element* existed and was deleted, 0 otherwise (the element was not there). */
int zsetDel(robj *zobj, sds ele) {if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {unsigned char *eptr;//ziplist先找到ele所在位置的指针eptrif ((eptr = zzlFind(zobj->ptr,ele,NULL)) != NULL) {//将该元素删除.ziplist删除时会resize,此处将删除之后ziplist的指针复值给zobj->ptrzobj->ptr = zzlDelete(zobj->ptr,eptr);return 1;}} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {zset *zs = zobj->ptr;dictEntry *de;double score;//skiplist现将zobj->ptr->dict相应的ele删除掉。此处并未真实删除//而是将ele所在的dictEntry返回de = dictUnlink(zs->dict,ele);if (de != NULL) {/* Get the score in order to delete from the skiplist later. *///通过dictEntry获取scorescore = *(double*)dictGetVal(de);/* Delete from the hash table and later from the skiplist.* Note that the order is important: deleting from the skiplist* actually releases the SDS string representing the element,* which is shared between the skiplist and the hash table, so* we need to delete from the skiplist as the final step. *///此处将dict中的key和value实际free掉dictFreeUnlinkedEntry(zs->dict,de);/* Delete from skiplist. *///从skiplist中删除元素.ele这个sds在hash和skiplist共享.从skiplist中删除时//会释放此sds,所以必须先删除dict中的元素再删除skiplist中的元素int retval = zslDelete(zs->zsl,score,ele,NULL);serverAssert(retval);//如果hash表中元素使用率小于10%,进行dict的resizeif (htNeedsResize(zs->dict)) dictResize(zs->dict);return 1;}} else {serverPanic("Unknown sorted set encoding");}return 0; /* No such element found. */
}