大家好，又见面了，我是你们的朋友全栈君。

redis6.0 源码学习（五）ziplist

一、数据结构

ziplist是经过特殊编码的双向链接列表，该列表具有很高的内存效率。 它存储字符串和整数值，其中整数被编码为实际整数，而不是一系列个字符。 它允许对列表的两侧进行push和pop操作且复杂度为O(1)。 但是由于每个操作都需要重新分配ziplist使用的内存，实际复杂度与ziplist使用的内存量有关。

下图是ziplist得示意图：

)

各个字段详细解释如下：

ziplist中的每个entry均包含两个信息：prevlen 和 encoding。prevlen 字段：存储前一个条目的长度，以便能够从后到前遍历列表。encoding表示entry类型，整数或字符串，对于字符串，还表示字符串有效负载的长度。 因此，完整的entry存储形式如下：

)

在存储较小的整数的时候，entry-data字段会不存在，因为也存储在encoding字段中了。

1、prevlen：针对prevlen的大小不同，prevlen占用的字段长度也不一样。

• prevlen < 254

  prevlen占用1个字节 8位

• prevlen > 254

  此时prevlen占用5个字节，第一个字节为 0xFE,后续四个字节为真正的长度。 0xFE表示是下一个entry一个大的值。

2、encoding 根据entry-data存入的是字符串还是整数具有不同的格式。具体如下：

• a：字符串

当字符串小于63字节时(2^6)，节点存为上图的第一种类型，高2位为00，低6位表示data的长度。

当字符串小于16383字节时(2^14)，节点存为上图的第二种类型，高2位为01，后续14位表示data的长度。

当字符串小于4294967296字节时(2^32)，节点存为上图的第三种类型，高2位为10，下一字节起连续32位表示data的长度。

• b：整数：整数节点的encoding的长度为8位，其中高2位用来区分整数节点和字符串节点（高2位为11时是整数节点），低6位用来区分整数节点的类型，具体如下：

二、代码解析

主要有以下接口，这里只分析几个，其他的大家可以自行去看源代码。

unsigned char *ziplistNew(void);
unsigned char *ziplistMerge(unsigned char **first, unsigned char **second);
unsigned char *ziplistPush(unsigned char *zl, unsigned char *s, unsigned int slen, int where);
unsigned char *ziplistIndex(unsigned char *zl, int index);
unsigned char *ziplistNext(unsigned char *zl, unsigned char *p);
unsigned char *ziplistPrev(unsigned char *zl, unsigned char *p);
unsigned int ziplistGet(unsigned char *p, unsigned char **sval, unsigned int *slen, long long *lval);
unsigned char *ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen);
unsigned char *ziplistDelete(unsigned char *zl, unsigned char **p);
unsigned char *ziplistDeleteRange(unsigned char *zl, int index, unsigned int num);
unsigned int ziplistCompare(unsigned char *p, unsigned char *s, unsigned int slen);
unsigned char *ziplistFind(unsigned char *p, unsigned char *vstr, unsigned int vlen, unsigned int skip);
unsigned int ziplistLen(unsigned char *zl);
size_t ziplistBlobLen(unsigned char *zl);

1、创建

可以对照图1 ziplist结构示意图 看这个创建的过程。

/* Create a new empty ziplist. */
unsigned char *ziplistNew(void) { 

unsigned int bytes = ZIPLIST_HEADER_SIZE+ZIPLIST_END_SIZE;  //<zlbytes>4字节<zltail>4字节<zllen>2字节<zlend>1字节，没有entry节点
unsigned char *zl = zmalloc(bytes); //申请内存
ZIPLIST_BYTES(zl) = intrev32ifbe(bytes);//zlbytes赋值
ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE);//zltail赋值
ZIPLIST_LENGTH(zl) = 0; //zllen赋值
zl[bytes-1] = ZIP_END; //zlend赋值
return zl;
}

2、查找

/* Find pointer to the entry equal to the specified entry. Skip 'skip' entries * between every comparison. Returns NULL when the field could not be found. */
//返回p节点之后data与vstr(长度是vlen)相等的节点，只找p节点之后每隔skip的节点
//时间复杂度 O(n)
unsigned char *ziplistFind(unsigned char *p, unsigned char *vstr, unsigned int vlen, unsigned int skip) { 

int skipcnt = 0;
unsigned char vencoding = 0;
long long vll = 0;
while (p[0] != ZIP_END) { 
  /* ZIP_END: Special "end of ziplist" entry. */
unsigned int prevlensize, encoding, lensize, len;
unsigned char *q;
ZIP_DECODE_PREVLENSIZE(p, prevlensize); //获取prevlen的字节数1还是5
/* 获取'ptr'中 entry encoding type 和 data length (字符串长度或者整数使用的bytes ) 。 * encoding变量： entry的encoding值 *lensize变量：encode entry所需要的bytes数 *len变量： hold the entry length. */        
ZIP_DECODE_LENGTH(p + prevlensize, encoding, lensize, len);
q = p + prevlensize + lensize; //当前节点的data
if (skipcnt == 0) { 

/* Compare current entry with specified entry */
if (ZIP_IS_STR(encoding)) { 
//判断当前节点是不是字符串节点
if (len == vlen && memcmp(q, vstr, vlen) == 0) { 
 //字符串比较
return p;
}
} else { 

/* Find out if the searched field can be encoded. Note that * we do it only the first time, once done vencoding is set * to non-zero and vll is set to the integer value. */
if (vencoding == 0) { 
 //这个代码块只会执行一次,判断vstr能够用整数表示 
/* 校验'vstr'指向的内容能否转换成整数，'vll'存这个整数值 'encoding'存编码格式 */
if (!zipTryEncoding(vstr, vlen, &vll, &vencoding)) { 

/* If the entry can't be encoded we set it to * UCHAR_MAX so that we don't retry again the next * time. */
vencoding = UCHAR_MAX;
}
/* Must be non-zero by now */
assert(vencoding); 
}
/* Compare current entry with specified entry, do it only * if vencoding != UCHAR_MAX because if there is no encoding * possible for the field it can't be a valid integer. */
if (vencoding != UCHAR_MAX) { 
 //entry地址的内容转换成整数
long long ll = zipLoadInteger(q, encoding);
if (ll == vll) { 

return p; //相等的话返回
}
}
}
/* Reset skip count */
skipcnt = skip;
} else { 

/* Skip entry */
skipcnt--;
}
/* Move to next entry */
p = q + len;
}
return NULL;
}

3、插入

/* Insert an entry at "p". */
unsigned char *ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen) { 

return __ziplistInsert(zl,p,s,slen);
}
/* Insert item at "p". */
unsigned char *__ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen) { 

size_t curlen = intrev32ifbe(ZIPLIST_BYTES(zl)), reqlen; // 当前长度和插入节点后需要的长度
unsigned int prevlensize, prevlen = 0; // 前置节点长度和编码该长度值所需的长度
size_t offset;
int nextdiff = 0;
unsigned char encoding = 0;
long long value = 123456789; /* initialized to avoid warning. Using a value that is easy to see if for some reason we use it uninitialized. */
zlentry tail;
/* Find out prevlen for the entry that is inserted. */
// 找出待插入节点的前置节点长度
// 如果p[0]不指向列表末端，说明列表非空，并且p指向其中一个节点
if (p[0] != ZIP_END) { 

// 获取前置节点p的长度和编码该长度需要的字节
ZIP_DECODE_PREVLEN(p, prevlensize, prevlen);
} else { 
        
unsigned char *ptail = ZIPLIST_ENTRY_TAIL(zl);
if (ptail[0] != ZIP_END) { 
// 如果p指向列表末端，表示列表为空
prevlen = zipRawEntryLength(ptail);
}
}
/* See if the entry can be encoded */
if (zipTryEncoding(s,slen,&value,&encoding)) { 
//尝试encoding成整数
/* 'encoding' is set to the appropriate integer encoding */
reqlen = zipIntSize(encoding); //获取编码长度
} else { 

/* 'encoding' is untouched, however zipStoreEntryEncoding will use the * string length to figure out how to encode it. */
reqlen = slen;
}
/* We need space for both the length of the previous entry and * the length of the payload. */
reqlen += zipStorePrevEntryLength(NULL,prevlen); // 获取前置节点的编码长度
reqlen += zipStoreEntryEncoding(NULL,encoding,slen); // 获取当前节点的编码长度
/* When the insert position is not equal to the tail, we need to * make sure that the next entry can hold this entry's length in * its prevlen field. */
// 只要不是列表的末端插入，需要计算出下一个元素保存本元素prevlen字段空间是否足够, 不够时计算出欠缺的差值 
// nextdiff大于0，那就说明需要对当前p指向的节点的header进行扩展
int forcelarge = 0;
nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0;
if (nextdiff == -4 && reqlen < 4) { 

nextdiff = 0;
forcelarge = 1;
}
/* Store offset because a realloc may change the address of zl. */
offset = p-zl;// 存储p相对于列表zl的偏移地址
zl = ziplistResize(zl,curlen+reqlen+nextdiff); // 重新分配空间，curlen当前列表的长度
p = zl+offset; // 重新获取p的值
/* Apply memory move when necessary and update tail offset. */
if (p[0] != ZIP_END) { 
 // 非表尾插入，需要重新计算表尾的偏移量
/* Subtract one because of the ZIP_END bytes */
memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff);/* 移动p原有位置和后面的内容到新的位置 */      
/* Encode this entry's raw length in the next entry. */   
if (forcelarge)
zipStorePrevEntryLengthLarge(p+reqlen,reqlen);
else
zipStorePrevEntryLength(p+reqlen,reqlen);
/* Update offset for tail */
// 更新列表尾相对于表头的偏移量，将新节点的长度算上
ZIPLIST_TAIL_OFFSET(zl) =
intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);
/* When the tail contains more than one entry, we need to take * "nextdiff" in account as well. Otherwise, a change in the * size of prevlen doesn't have an effect on the *tail* offset. */
// 如果新节点后面有多个节点，那么表尾的偏移量需要算上nextdiff的值
zipEntry(p+reqlen, &tail);
if (p[reqlen+tail.headersize+tail.len] != ZIP_END) { 

ZIPLIST_TAIL_OFFSET(zl) =
intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
}
} else { 

// 表尾插入，直接计算偏移量
/* This element will be the new tail. */
ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(p-zl);
}
/* When nextdiff != 0, the raw length of the next entry has changed, so * we need to cascade the update throughout the ziplist */
// 当nextdiff不为0时，表示需要新节点的后继节点对头部进行扩展
//说明下一个元素需要扩展空间存放prevlen字段, 由于下一个元素空间变大, 有可能引起下下一个元素空间需要扩展, 下面函数检测后面元素, 并在需要时重置元素prevlen长度
if (nextdiff != 0) { 

// 需要对p所指向的header进行扩展更新
// 有可能会引起连锁更新
offset = p-zl;
zl = __ziplistCascadeUpdate(zl,p+reqlen);
p = zl+offset;
}
/* Write the entry */
p += zipStorePrevEntryLength(p,prevlen); // 将新节点前置节点的长度写入新节点的header
p += zipStoreEntryEncoding(p,encoding,slen); // 将新节点的值长度写入新节点的header
if (ZIP_IS_STR(encoding)) { 
 // 写入节点值
memcpy(p,s,slen);
} else { 

zipSaveInteger(p,value,encoding);
}
ZIPLIST_INCR_LENGTH(zl,1);  // 更新列表节点计数
return zl;
}

三、总结

• ziplist是为节省内存空间而生的。
• ziplist是一个为Redis专门提供的底层数据结构之一，本身可以有序也可以无序。当作hash的底层实现时，节点之间没有顺序；当作为zset的底层实现时，节点之间会按照大小顺序排列。