【Redis源码分析】Redis的压缩列表ZipList

时间 2019-11-10

标签 Redis源码分析 redis 压缩列表 ziplist 栏目 Redis 繁體版

原文原文链接

本文将介绍Redis中一种重要的数据结构，这种数据结构是为了节省内存而设计的，这就是压缩列表（ZipList）。redis

一、简介

压缩列表（ziplist）本质上就是一个字节数组，是Redis为了节约内存而设计的一种线性数据结构，能够包含任意多个元素，每一个元素能够是一个字节数组或一个整数。
Redis的有序集合、哈希以及列表都直接或者间接使用了压缩列表。当有序集合或哈希的元素数目比较少，且元素都是短字符串时，Redis便使用压缩列表做为其底层数据存储方式。列表使用快速链表（quicklist）数据结构存储，而快速链表就是双向链表与压缩列表的组合。
例如，使用下面命令建立一个哈希键并查看其编码：shell

127.0.0.1:6379> hmset person name zhangsan gender 1 age 22
OK
127.0.0.1:6379> object encoding person
"ziplist"

二、数据存储

2.1 编码

Redis使用字节数组表示一个压缩列表，字节数组逻辑划分为多个字段，如图所示：segmentfault

各字段含义以下：数组

一、zlbytes：压缩列表的字节长度，占4个字节，所以压缩列表最长(2^32)-1字节；
二、zltail：压缩列表尾元素相对于压缩列表起始地址的偏移量，占4个字节；
三、zllen：压缩列表的元素数目，占两个字节；那么当压缩列表的元素数目超过(2^16)-1怎么处理呢？此时经过zllen字段没法得到压缩列表的元素数目，必须遍历整个压缩列表才能获取到元素数目；
四、entryX：压缩列表存储的若干个元素，能够为字节数组或者整数；entry的编码结构后面详述；
五、zlend：压缩列表的结尾，占一个字节，恒为0xFF。

假设char * zl指向压缩列表首地址，Redis经过如下宏定义实现了压缩列表各个字段的操做存取：缓存

//zl指向zlbytes字段
#define ZIPLIST_BYTES(zl)       (*((uint32_t*)(zl)))

//zl+4指向zltail字段
#define ZIPLIST_TAIL_OFFSET(zl) (*((uint32_t*)((zl)+sizeof(uint32_t))))

//zl+zltail指向尾元素首地址；intrev32ifbe使得数据存取统一按照小端法
#define ZIPLIST_ENTRY_TAIL(zl)   ((zl)+intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))) 

//zl+8指向zllen字段
#define ZIPLIST_LENGTH(zl)   (*((uint16_t*)((zl)+sizeof(uint32_t)*2)))

//压缩列表最后一个字节即为zlend字段
#define ZIPLIST_ENTRY_END(zl)   ((zl)+intrev32ifbe(ZIPLIST_BYTES(zl))-1)

了解了压缩列表的基本结构，咱们能够很容易得到压缩列表的字节长度，元素数目等，那么如何遍历压缩列表的全部元素呢？咱们已经知道对于每个entry元素，存储的多是字节数组或整数值；那么对任一个元素，咱们如何判断存储的是什么类型？对于字节数组，咱们又如何获取字节数组的长度？
回答这些问题以前，须要先看看压缩列表元素的编码结构，如图所示：数据结构

previous_entry_length字段表示前一个元素的字节长度，占1个或者5个字节；当前一个元素的长度小于254字节时，previous_entry_length字段用一个字节表示；当前一个元素的长度大于等于254字节时，previous_entry_length字段用5个字节来表示；而这时候previous_entry_length的第一个字节是固定的标志0xFE，后面4个字节才真正表示前一个元素的长度；假设已知当前元素的首地址为p，那么（p-previous_entry_length）就是前一个元素的首地址，从而实现压缩列表从尾到头的遍历；
encoding字段表示当前元素的编码，即content字段存储的数据类型（整数或者字节数组），数据内容存储在content字段；为了节约内存，encoding字段一样是可变长度，编码如表-1所示：
<center>表-1 压缩列表元素编码表格</center>curl

encoding编码	encoding长度	content类型
00 bbbbbb（6比特表示content长度）	1字节	最大长度63的字节数组
01 bbbbbb xxxxxxxx（14比特表示content长度）	2字节	最大长度(2^14)-1字节最大长度的字节数组
10 __ aaaaaaaa bbbbbbbb cccccccc dddddddd（32比特表示content长度）	5字节	最大长度(2^32)-1的字节数组
11 00 0000	1字节	int16整数
11 01 0000	1字节	int32整数
11 10 0000	1字节	int64整数
11 11 0000	1字节	24比特整数
11 11 1110	1字节	8比特整数
11 11 xxxx	1字节	没有content字段；xxxx表示0~12之间的整数

能够看出，根据encoding字段第一个字节的前2个比特，能够判断content字段存储的是整数，或者字节数组（以及字节数组最大长度）；当content存储的是字节数组时，后续字节标识字节数组的实际长度；当content存储的是整数时，根据第三、4比特才能判断整数的具体类型；而当encoding字段标识当前元素存储的是0~12的当即数时，数据直接存储在encoding字段的最后4个比特，此时没有content字段。参照encoding字段的编码表格，Redis预约义了如下常量：函数

#define ZIP_STR_06B (0 << 6)
#define ZIP_STR_14B (1 << 6)
#define ZIP_STR_32B (2 << 6)
#define ZIP_INT_16B (0xc0 | 0<<4)
#define ZIP_INT_32B (0xc0 | 1<<4)
#define ZIP_INT_64B (0xc0 | 2<<4)
#define ZIP_INT_24B (0xc0 | 3<<4)
#define ZIP_INT_8B 0xfe

2.2 结构体entry

2.1节分析了压缩列表的底层存储结构。可是咱们发现对于任意的压缩列表元素，获取前一个元素的长度，判断存储的数据类型，获取数据内容，都须要通过复杂的解码运算才行，那么解码后的结果应该被缓存起来，为此定义告终构体zlentry，用于表示解码后的压缩列表元素：ui

typedef struct zlentry {
    unsigned int prevrawlensize;
    unsigned int prevrawlen;
     
    unsigned int lensize;                                 
    unsigned int len;
            
    unsigned int headersize; 
    
    unsigned char encoding;
      
    unsigned char *p;           
} zlentry;

咱们看到结构体定义了7个字段，而2.1节显示每一个元素只包含3个字段。回顾压缩列表元素的编码结构，可变因素实际上不止三个；previous_entry_length字段的长度（字段prevrawlensize表示）、previous_entry_length字段存储的内容（字段prevrawlen表示）、encoding字段的长度（字段lensize表示）、encoding字段的内容（字段len表示数据内容长度，字段encoding表示数据类型）、和当前元素首地址（字段p表示）。而headersize字段则表示当前元素的首部长度，即previous_entry_length字段长度与encoding字段长度之和。
函数zipEntry用来解码压缩列表的元素，存储于zlentry结构体：编码

void zipEntry(unsigned char *p, zlentry *e) {
    ZIP_DECODE_PREVLEN(p, e->prevrawlensize, e->prevrawlen);
    ZIP_DECODE_LENGTH(p + e->prevrawlensize, e->encoding, e->lensize, e->len);
    e->headersize = e->prevrawlensize + e->lensize;
    e->p = p;
}

解码过程主要能够分为如下两个步骤：

1) 解码previous_entry_length字段，此时入参ptr指向元素首地址；

#define ZIP_BIG_PREVLEN 254

#define ZIP_DECODE_PREVLEN(ptr, prevlensize, prevlen) do {     \
    if ((ptr)[0] < ZIP_BIG_PREVLEN) {                          \
        (prevlensize) = 1;                                     \
        (prevlen) = (ptr)[0];                                  \
    } else {                                                   \
        (prevlensize) = 5;                                     \
        memcpy(&(prevlen), ((char*)(ptr)) + 1, 4);             \
        memrev32ifbe(&prevlen);                                \
    }                                                          \
} while(0);

2) 解码encoding字段逻辑，此时ptr指向元素首地址偏移previous_entry_length字段长度的位置：

#define ZIP_STR_MASK 0xc0

#define ZIP_DECODE_LENGTH(ptr, encoding, lensize, len) do {    \
    (encoding) = (ptr[0]);                                     \
    // ptr[0]小于11000000说明是字节数组，前两个比特为编码
    if ((encoding) < ZIP_STR_MASK) (encoding) &= ZIP_STR_MASK; \
    if ((encoding) < ZIP_STR_MASK) {                           \
        if ((encoding) == ZIP_STR_06B) {                       \
            (lensize) = 1;                                     \
            (len) = (ptr)[0] & 0x3f;                           \
        } else if ((encoding) == ZIP_STR_14B) {                \
            (lensize) = 2;                                     \
            (len) = (((ptr)[0] & 0x3f) << 8) | (ptr)[1];       \
        } else if ((encoding) == ZIP_STR_32B) {                \
            (lensize) = 5;                                     \
            (len) = ((ptr)[1] << 24) |                         \
                    ((ptr)[2] << 16) |                         \
                    ((ptr)[3] <<  8) |                         \
                    ((ptr)[4]);                                \
        } else {                                               \
            panic("Invalid string encoding 0x%02X", (encoding));\
        }                                                      \
    } else {                                                   \
        (lensize) = 1;                                         \
        (len) = zipIntSize(encoding);                          \
    }                                                          \
} while(0);

字节数组只根据ptr[0]的前2个比特便可判类型，而判断整数类型须要ptr[0]的前4个比特，代码以下：

unsigned int zipIntSize(unsigned char encoding) {
    switch(encoding) {
    case ZIP_INT_8B:  return 1;
    case ZIP_INT_16B: return 2;
    case ZIP_INT_24B: return 3;
    case ZIP_INT_32B: return 4;
    case ZIP_INT_64B: return 8;
    }
    
    // 4比特当即数
    if (encoding >= ZIP_INT_IMM_MIN && encoding <= ZIP_INT_IMM_MAX)     
        return 0; 
    panic("Invalid integer encoding 0x%02X", encoding);
    return 0;
}

三、基本操做

3.1 建立压缩列表

建立压缩列表的API定义以下，函数无输入参数，返回参数为压缩列表首地址：

unsigned char *ziplistNew(void);
建立空的压缩列表，只须要分配初始存储空间（11=4+4+2+1个字节），并对zlbytes、zltail、zllen和zlend字段初始化便可。
unsigned char *ziplistNew(void) {
    //ZIPLIST_HEADER_SIZE = zlbytes + zltail + zllen;
    unsigned int bytes = ZIPLIST_HEADER_SIZE+1;        
    unsigned char *zl = zmalloc(bytes);
 
    ZIPLIST_BYTES(zl) = intrev32ifbe(bytes);
    ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE);
    ZIPLIST_LENGTH(zl) = 0;
    
    //结尾标识0XFF
    zl[bytes-1] = ZIP_END;             
    return zl;
}

3.2 插入元素

压缩列表插入元素的API定义以下，函数输入参数zl表示压缩列表首地址，p指向新元素的插入位置，s表示数据内容，slen表示数据长度，返回参数为压缩列表首地址。

unsigned char *ziplistInsert(unsigned char *zl, unsigned char *p,  
                                     unsigned char *s, unsigned int slen);

插入元素时，能够简要分为三个步骤：第一步须要将元素内容编码为压缩列表的元素，第二步从新分配空间，第三步拷贝数据。下面分别讨论每一个步骤的实现逻辑。

1) 编码

编码即计算previous_entry_length字段、encoding字段和content字段的内容。如何获取前一个元素的长度呢？这时候就须要根据插入元素的位置分状况讨论了，如图所示：

当压缩列表为空插入位置为P0时，此时不存在前一个元素，即前一个元素的长度为0；
当插入位置为P1时，此时须要获取entryX元素的长度，而entryX+1元素的previous_entry_length字段存储的就是entryX元素的长度，比较容易获取；
当插入位置为P2时，此时须要获取entryN元素的长度，entryN是压缩列表的尾元素，计算其元素长度须要将其三个字段长度相加，函数实现以下：

unsigned int zipRawEntryLength(unsigned char *p) {
    unsigned int prevlensize, encoding, lensize, len;
    ZIP_DECODE_PREVLENSIZE(p, prevlensize);
    ZIP_DECODE_LENGTH(p + prevlensize, encoding, lensize, len);
    return prevlensize + lensize + len;
}

其中ZIP_DECODE_PREVLENSIZE和ZIP_DECODE_LENGTH在2.2节已经讲过，这里再也不赘述。
encoding字段标识的是当前元素存储的数据类型以及数据长度，编码时首先会尝试将数据内容解析为整数，若是解析成功则按照压缩列表整数类型编码存储，解析失败的话按照压缩列表字节数组类型编码存储。

if (zipTryEncoding(s,slen,&value,&encoding)) {
    reqlen = zipIntSize(encoding);
} else {
    reqlen = slen;
}
 
reqlen += zipStorePrevEntryLength(NULL,prevlen);
reqlen += zipStoreEntryEncoding(NULL,encoding,slen);

程序首先尝试按照整数解析新添加元素的数据内容，数值存储在变量value，编码存储在变量encoding。若是解析成功，还须要计算整数所占字节数。
变量reqlen最终存储的是当前元素所需空间大小，初始赋值为元素content字段所需空间大小，再累加previous_entry_length所需空间大小与encoding字段所需空间大小。

2) 从新分配空间

因为新插入元素，压缩列表所需空间增大，所以须要从新分配存储空间。那么空间大小是不是添加元素前的压缩列表长度与新添加元素元素长度之和呢？并不彻底是，如图中所示的例子。
插入元素前，entryX元素长度为128字节，entryX+1元素的previous_entry_length字段占1个字节；添加元素entryNEW元素，元素长度为1024字节，此时entryX+1元素的previous_entry_length字段须要占5个字节；即压缩列表的长度不只仅是增长了1024字节，还有entryX+1元素扩展的4字节。咱们很容易知道，entryX+1元素长度可能增长4字节，也可能减少4字节，也可能不变。而因为从新分配空间，新元素插入的位置指针P会失效，所以须要预先计算好指针P相对于压缩列表首地址的偏移量，待分配空间以后再偏移便可。

size_t curlen = intrev32ifbe(ZIPLIST_BYTES(zl));
 
int forcelarge = 0;
nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0;
if (nextdiff == -4 && reqlen < 4) {
    nextdiff = 0;
    forcelarge = 1;
}
 
//存储偏移量
offset = p-zl;
//调用realloc从新分配空间
zl = ziplistResize(zl,curlen+reqlen+nextdiff);
//从新偏移到插入位置P
p = zl+offset;

那么nextdiff与forcelarge在这里有什么用呢？分析ziplistResize函数的3个输入参数，curlen表示插入元素前压缩列表的长度，reqlen表示插入元素元素的长度，而nextdiff表示的是entryX+1元素长度的变化，取值可能为0（长度不变）、4（长度增长4）和-4（长度减少4）。咱们再思考下，当nextdiff等于-4，而reqlen小于4时会发生什么呢？没错，插入元素致使压缩列表所需空间减小了，即函数ziplistResize底层调用realloc从新分配的空间小于指针zl指向的空间。这可能会存在问题，咱们都知道realloc从新分配空间时，返回的地址可能不变，当从新分配的空间大小反而减小时，realloc底层实现可能会将多余的空间回收，此时可能会致使数据的丢失。所以须要避免这种状况的发生，即从新赋值nextdiff等于0，同时使用forcelarge标记这种状况。
能够再思考下，nextdiff等于-4时，reqlen会小于4吗？答案是可能的，连锁更新可能会致使这种状况的发生。连锁更新将在第4节介绍。

3) 数据拷贝

从新分配空间以后，须要将位置P后的元素移动到指定位置，将新元素插入到位置P。咱们假设entryX+1元素的长度增长4（即nextdiff等于4），此时数据拷贝示意图如图所示：

从图中能够看到，位置P后的全部元素都须要移动，移动的偏移量是插入元素entryNew的长度，移动的数据块长度是位置P后全部元素长度之和再加上nextdiff的值，数据移动以后还须要更新entryX+1元素的previous_entry_length字段。

memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff); 
//更新entryX+1元素的previous_entry_length字段字段
if (forcelarge)
    //entryX+1元素的previous_entry_length字段依然占5个字节;
    //可是entryNEW元素长度小于4字节
    zipStorePrevEntryLengthLarge(p+reqlen,reqlen);
else
    zipStorePrevEntryLength(p+reqlen,reqlen);
 
//更新zltail字段
ZIPLIST_TAIL_OFFSET(zl) =
    intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);
 
zipEntry(p+reqlen, &tail);
if (p[reqlen+tail.headersize+tail.len] != ZIP_END) {
    ZIPLIST_TAIL_OFFSET(zl) =
        intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
}
 
//更新zllen字段
ZIPLIST_INCR_LENGTH(zl,1);

思考一下，第一次更新尾元素偏移量以后，为何指向的元素可能不是尾元素呢？很显然，当entryX+1元素就是尾元素时，只须要更新一次尾元素的偏移量；可是当entryX+1元素不知尾元素，且entryX+1元素长度发生了改变，此时尾元素偏移量还须要加上nextdiff的值。

3.3 删除元素

压缩列表删除元素的API定义以下，函数输入参数zl指向压缩列表首地址，*p指向待删除元素的首地址（参数p同时能够做为输出参数），返回参数为压缩列表首地址。

unsigned char *ziplistDelete(unsigned char *zl, unsigned char **p);

ziplistDelete函数只是简单调用底层__ziplistDelete函数实现删除功能；__ziplistDelete函数能够同时删除多个元素，输入参数p指向的是首个删除元素的首地址，num表示待删除元素数目。

unsigned char *ziplistDelete(unsigned char *zl, unsigned char **p) {
    size_t offset = *p-zl;
    zl = __ziplistDelete(zl,*p,1);
    *p = zl+offset;
    return zl;
}

删除元素一样能够简要分为三个步骤：第一步计算待删除元素总长度，第二步数据拷贝，第三步从新分配空间。下面分别讨论每一个步骤的实现逻辑。

1) 计算待删除元素总长度，其中zipRawEntryLength函数在3.2节已经讲过，这里再也不详述；

//解码第一个待删除元素
zipEntry(p, &first);
 
//遍历全部待删除元素，同时指针p向后偏移
for (i = 0; p[0] != ZIP_END && i < num; i++) {
    p += zipRawEntryLength(p);
    deleted++;
}
//totlen即为待删除元素总长度
totlen = p-first.p;

2) 数据拷贝；

第一步骤计算完成以后，指针first与指针p之间的元素都是待删除的。很显然，当指针p刚好指向zlend字段，再也不须要数据的拷贝了，只须要更新尾节点的偏移量便可。下面分析另一种状况，即指针p指向的是某一个元素而不是zlend字段。
分析相似于3.2节插入元素。删除元素时，压缩列表所需空间减小，减小的量是否仅仅是待删除元素总长度呢？答案一样是否认的，举个简单的例子，下图是通过第一步骤计算以后的示意图：

删除元素entryX+1到元素entryN-1之间的N-X-1个元素，元素entryN-1的长度为12字节，所以元素entryN的previous_entry_length字段占1个字节；删除这些元素以后，entryX称为了entryN的前一个元素，元素entryX的长度为512字节，所以元素entryN的previous_entry_length字段须要占5个字节。即删除元素以后的压缩列表的总长度，还与entryN元素长度的变化量有关。

//计算元素entryN长度的变化量
nextdiff = zipPrevLenByteDiff(p,first.prevrawlen);
 
//更新元素entryN的previous_entry_length字段
p -= nextdiff;
zipStorePrevEntryLength(p,first.prevrawlen);
 
//更新zltail
ZIPLIST_TAIL_OFFSET(zl) =
    intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))-totlen);
 
zipEntry(p, &tail);
if (p[tail.headersize+tail.len] != ZIP_END) {
    ZIPLIST_TAIL_OFFSET(zl) =
       intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
}
 
//数据拷贝
memmove(first.p,p,
    intrev32ifbe(ZIPLIST_BYTES(zl))-(p-zl)-1);

与3.2节插入元素更新zltail字段相同，当entryX+1元素就是尾元素时，只须要更新一次尾元素的偏移量；可是当entryX+1元素不是尾元素时，且entryX+1元素长度发生了改变，此时尾元素偏移量还须要加上nextdiff的值。

3) 从新分配空间

逻辑与3.2节插入元素逻辑基本相似，这里就再也不详述。代码以下：

offset = first.p-zl;
zl = ziplistResize(zl, intrev32ifbe(ZIPLIST_BYTES(zl))-totlen+nextdiff);
p = zl+offset;
ZIPLIST_INCR_LENGTH(zl,-deleted);

思考一下：在3.2节咱们提到，调用ziplistResize函数从新分配空间时，若是从新分配的空间小于指针zl指向的空间大小时，可能会出现问题。而这里因为是删除元素，压缩列表的长度确定是减小的。为何又能这样使用呢？
根本缘由在于删除元素时，咱们是先拷贝数据，再从新分配空间，即调用ziplistResize函数时，多余的那部分空间存储的数据已经被拷贝了，此时回收这部分空间并不会形成数据的丢失。

3.5 遍历压缩列表

遍历就是从头至尾（前向遍历）或者从尾到头（后向遍历）访问压缩列表中的每个元素。压缩列表的遍历API定义以下，函数输入参数zl指向压缩列表首地址，p指向当前访问元素的首地址；ziplistNext函数返回后一个元素的首地址，ziplistPrev返回前一个元素的首地址。

//后向遍历
unsigned char *ziplistNext(unsigned char *zl, unsigned char *p);
//前向遍历
unsigned char *ziplistPrev(unsigned char *zl, unsigned char *p);

咱们已经知道压缩列表每一个元素的previous_entry_length字段存储的是前一个元素的长度，所以压缩列表的前向遍历相对简单，表达式(p-previous_entry_length)便可获取前一个元素的首地址，这里不作详述。后向遍历时，须要解码当前元素，计算当前元素长度，才能获取后一个元素首地址；ziplistNext函数实现以下：

unsigned char *ziplistNext(unsigned char *zl, unsigned char *p) {
    //zl参数无用；这里只是为了不警告
    ((void) zl);
 
    if (p[0] == ZIP_END) {
        return NULL;
    }
 
    p += zipRawEntryLength(p);
    if (p[0] == ZIP_END) {
        return NULL;
    }
 
    return p;
}

四、连锁更新

以下图所示，删除压缩列表zl1位置P1的元素entryX，或者在压缩列表zl2位置P2插入元素entryY，此时会出现什么状况呢？

压缩列表zl1，元素entryX以后的全部元素entryX+一、entryX+2等长度都是253字节，显然这些元素的previous_entry_length字段的长度都是1字节。当删除元素entryX时，元素entryX+1的前驱节点改成元素entryX-1，长度为512字节，此时元素entryX+1的previous_entry_length字段须要5字节才能存储元素entryX-1的长度，则元素entryX+1的长度须要扩展至257字节；而因为元素entryX+1长度的增长，元素entryX+2的previous_entry_length字段一样须要改变。以此类推，因为删除了元素entryX，以后的全部元素entryX+一、entryX+2等长度都必须扩展，而每次元素的扩展都将致使从新分配内存，效率是很低下的。压缩列表zl2，插入元素entryY一样会产生上面的问题。
上面的状况称之为连锁更新。从上面分析能够看出，连锁更新会致使屡次从新分配内存以及数据拷贝，效率是很低下的。可是出现这种状况的几率是很低的，所以对于删除元素与插入元素的操做，redis并无为了不连锁更新而采起措施。redis只是在删除元素与插入元素操做的末尾，检查是否须要更新后续元素的previous_entry_length字段，其实现函数_ziplistCascadeUpdate，主要逻辑以下图所示：

五、总结

本文首先介绍了压缩列表的编码与数据结构，随后介绍了压缩列表的基本操做：建立压缩列表、插入元素、删除元素与遍历，最后分析了压缩列表连锁更新的出现以及解决方案。