zset底层存储结构
zset底层的存储结构包括ziplist或skiplist，在同时满足以下两个条件的时候使用ziplist，其他时候使用skiplist，两个条件如下：
- 有序集合保存的元素数量小于128个
- 有序集合保存的所有元素的长度小于64字节
当ziplist作为zset的底层存储结构时候，每个集合元素使用两个紧挨在一起的压缩列表节点来保存，第一个节点保存元素的成员，第二个元素保存元素的分值。
当skiplist作为zset的底层存储结构的时候，使用skiplist按序保存元素及分值，使用dict来保存元素和分值的映射关系。
ziplist数据结构
ziplist作为zset的存储结构时，格式如下图，细节就不多说了，我估计大家都看得懂，紧挨着的是元素memeber和分值socore，整体数据是有序格式。
zset ziplist结构
skiplist数据结构
skiplist作为zset的存储结构，整体存储结构如下图，核心点主要是包括一个dict对象和一个skiplist对象。dict保存key/value，key为元素，value为分值；skiplist保存的有序的元素列表，每个元素包括元素和分值。两种数据结构下的元素指向相同的位置。
zset skiplist结构
skiplist的源码格式
zset包括dict和zskiplist两个数据结构，其中dict的保存key/value，便于通过key(元素)获取score(分值)。zskiplist保存有序的元素列表，便于执行range之类的命令。
/*
* 有序集合
*/
typedef struct zset {
// 字典，键为成员，值为分值
// 用于支持 O(1) 复杂度的按成员取分值操作
dict *dict;
// 跳跃表，按分值排序成员
// 用于支持平均复杂度为 O(log N) 的按分值定位成员操作
// 以及范围操作
zskiplist *zsl;
} zset;
zskiplist作为skiplist的数据结构，包括指向头尾的header和tail指针，其中level保存的是skiplist的最大的层数。
/*
* 跳跃表
*/
typedef struct zskiplist {
// 表头节点和表尾节点
struct zskiplistNode *header, *tail;
// 表中节点的数量
unsigned long length;
// 表中层数最大的节点的层数
int level;
} zskiplist;
skiplist跳跃列表中每个节点的数据格式，每个节点有保存数据的robj指针，分值score字段，后退指针backward便于回溯，zskiplistLevel的数组保存跳跃列表每层的指针。
/*
* 跳跃表节点
*/
typedef struct zskiplistNode {
// 成员对象
robj *obj;
// 分值
double score;
// 后退指针
struct zskiplistNode *backward;
// 层
struct zskiplistLevel {
// 前进指针
struct zskiplistNode *forward;
// 跨度
unsigned int span;
} level[];
} zskiplistNode;
zset存储过程
zset的添加过程我们以zadd的操作作为例子进行分析，整个过程如下：
- 解析参数得到每个元素及其对应的分值
- 查找key对应的zset是否存在不存在则创建
- 如果存储格式是ziplist，那么在执行添加的过程中我们需要区分元素存在和不存在两种情况，存在情况下先删除后添加；不存在情况下则添加并且需要考虑元素的长度是否超出限制或实际已有的元素个数是否超过最大限制进而决定是否转为skiplist对象。
- 如果存储格式是skiplist，那么在执行添加的过程中我们需要区分元素存在和不存在两种情况，存在的情况下先删除后添加，不存在情况下那么就直接添加，在skiplist当中添加完以后我们同时需要更新dict的对象。
void zaddGenericCommand(redisClient *c, int incr) {
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 = (c->argc-2)/2;
int added = 0, updated = 0;
// 输入的 score - member 参数必须是成对出现的
if (c->argc % 2) {
addReply(c,shared.syntaxerr);
return;
}
// 取出所有输入的 score 分值
scores = zmalloc(sizeof(double)*elements);
for (j = 0; j < elements; j++) {
if (getDoubleFromObjectOrReply(c,c->argv[2+j*2],&scores[j],NULL)
!= REDIS_OK) goto cleanup;
}
// 取出有序集合对象
zobj = lookupKeyWrite(c->db,key);
if (zobj == NULL) {
// 有序集合不存在，创建新有序集合
if (server.zset_max_ziplist_entries == 0 ||
server.zset_max_ziplist_value < sdslen(c->argv[3]->ptr))
{
zobj = createZsetObject();
} else {
zobj = createZsetZiplistObject();
}
// 关联对象到数据库
dbAdd(c->db,key,zobj);
} else {
// 对象存在，检查类型
if (zobj->type != REDIS_ZSET) {
addReply(c,shared.wrongtypeerr);
goto cleanup;
}
}
// 处理所有元素
for (j = 0; j < elements; j++) {
score = scores[j];
// 有序集合为 ziplist 编码
if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
unsigned char *eptr;
// 查找成员
ele = c->argv[3+j*2];
if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) {
// 成员已存在
// ZINCRYBY 命令时使用
if (incr) {
score += curscore;
if (isnan(score)) {
addReplyError(c,nanerr);
goto cleanup;
}
}
// 执行 ZINCRYBY 命令时，
// 或者用户通过 ZADD 修改成员的分值时执行
if (score != curscore) {
// 删除已有元素
zobj->ptr = zzlDelete(zobj->ptr,eptr);
// 重新插入元素
zobj->ptr = zzlInsert(zobj->ptr,ele,score);
// 计数器
server.dirty++;
updated++;
}
} else {
// 元素不存在，直接添加
zobj->ptr = zzlInsert(zobj->ptr,ele,score);
// 查看元素的数量，
// 看是否需要将 ZIPLIST 编码转换为有序集合
if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries)
zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
// 查看新添加元素的长度
// 看是否需要将 ZIPLIST 编码转换为有序集合
if (sdslen(ele->ptr) > server.zset_max_ziplist_value)
zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
server.dirty++;
added++;
}
// 有序集合为 SKIPLIST 编码
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
zskiplistNode *znode;
dictEntry *de;
// 编码对象
ele = c->argv[3+j*2] = tryObjectEncoding(c->argv[3+j*2]);
// 查看成员是否存在
de = dictFind(zs->dict,ele);
if (de != NULL) {
// 成员存在
// 取出成员
curobj = dictGetKey(de);
// 取出分值
curscore = *(double*)dictGetVal(de);
// ZINCRYBY 时执行
if (incr) {
score += curscore;
if (isnan(score)) {
addReplyError(c,nanerr);
goto cleanup;
}
}
// 执行 ZINCRYBY 命令时，
// 或者用户通过 ZADD 修改成员的分值时执行
if (score != curscore) {
// 删除原有元素
redisAssertWithInfo(c,curobj,zslDelete(zs->zsl,curscore,curobj));
// 重新插入元素
znode = zslInsert(zs->zsl,score,curobj);
incrRefCount(curobj); /* Re-inserted in skiplist. */
// 更新字典的分值指针
dictGetVal(de) = &znode->score; /* Update score ptr. */
server.dirty++;
updated++;
}
} else {
// 元素不存在，直接添加到跳跃表
znode = zslInsert(zs->zsl,score,ele);
incrRefCount(ele); /* Inserted in skiplist. */
// 将元素关联到字典
redisAssertWithInfo(c,NULL,dictAdd(zs->dict,ele,&znode->score) == DICT_OK);
incrRefCount(ele); /* Added to dictionary. */
server.dirty++;
added++;
}
} else {
redisPanic("Unknown sorted set encoding");
}
}
if (incr) /* ZINCRBY */
addReplyDouble(c,score);
else /* ZADD */
addReplyLongLong(c,added);
cleanup:
zfree(scores);
if (added || updated) {
signalModifiedKey(c->db,key);
notifyKeyspaceEvent(REDIS_NOTIFY_ZSET,
incr ? "zincr" : "zadd", key, c->db->id);
}
}
