Redis源码剖析--压缩列表

柔光的暖阳◎ 2022-06-08 00:38 237阅读 0赞

压缩列表（ziplist）是列表键和哈希键的底层实现之一。  
Redis的列表键，哈希键，有序集合的底层实现都用到了ziplist。

当列表键中包含比较少的元素，并且元素都是数字或者比较小的字符串的时候， redis会用压缩列表来作为列表键的底层实现。

当哈希键的键和值都是比较小的整数或者较短的字符的时候，也是用压缩列表来作为底层实现。 因为压缩列表也能够节省内存。

## 压缩列表结构 ##

压缩列表的结构如下：

![压缩列表结构][SouthEast]

列表头包括三部分内容，分别是zlbytes，zltail，zllen

*  zlbytes： 记录整个压缩列表占用的内存字节数：在对压缩列表进行内存重分配， 或者计算 zlend 的位置时使用。
 *  zltail：记录压缩列表表尾节点距离压缩列表的起始地址有多少字节： 通过这个偏移量，程序无须遍历整个压缩列表就可以确定表尾节点的地址。
 *  zllen：记录了压缩列表包含的节点数量： 当这个属性的值小于 UINT16\_MAX （65535）时， 这个属性的值就是压缩列表包含节点的数量； 当这个值等于 UINT16\_MAX 时， 节点的真实数量需要遍历整个压缩列表才能计算得出。

压缩列表中间一次保存着各个列表项entry。

压缩列表尾部的zlend则表示压缩列表结束，其值固定为0xFF。

### 压缩列表结点 ###

先看结点的数据结构：

typedef struct zlentry {
        unsigned int prevrawlensize, prevrawlen; // 前置节点长度和编码所需长度
        unsigned int lensize, len; // 当前节点长度和编码所需长度
        unsigned int headersize; // 头的大小
        unsigned char encoding; // 编码类型
        unsigned char *p; // 数据部分
    } zlentry;

每个压缩列表节点都由 previous\_entry\_length 、 encoding 、 content 三个部分组成。

#### previous\_entry\_length ####

节点的 previous\_entry\_length 记录了压缩列表中前一个节点的长度。

previous\_entry\_length 属性的长度可以是 1 字节或者 5 字节：

*  如果前一节点的长度小于 254 字节， 那么 previous\_entry\_length 属性的长度为 1 字节： 前一节点的长度就保存在这一个字节里面。
 *  如果前一节点的长度大于等于 254 字节， 那么 previous\_entry\_length 属性的长度为 5 字节： 其中属性的第一字节会被设置为 0xFE （十进制值 254）， 而之后的四个字节则用于保存前一节点的长度。

压缩列表zltail和previous\_entry\_length的存在，我们能够轻松得到一个列表的尾部，然后从尾部实现向前遍历整个压缩列表。

#### encoding ####

压缩列表能够保存字节数组和整数，当读取压缩列表的时候，如何区分当前的结点存储的是字节数组还是整数呢，就需要靠encoding字段来判断。

##### 1、字节数组 #####

保存字节数组的时候，encoding字段可以是一字节、两字节或者五字节长， 值的最高位为 00 、 01 或者 10 ，数组的长度由编码除去最高两位之后的其他位记录。

<table> 
 <thead> 
  <tr> 
   <th>编码</th> 
   <th>编码长度</th> 
   <th>content 属性保存的值</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>00bbbbbb</td> 
   <td>1 字节</td> 
   <td>长度小于等于 63 字节的字节数组。</td> 
  </tr> 
  <tr> 
   <td>01bbbbbb xxxxxxxx</td> 
   <td>2 字节</td> 
   <td>长度小于等于 16383 字节的字节数组。</td> 
  </tr> 
  <tr> 
   <td>10______ aaaaaaaa bbbbbbbb cccccccc dddddddd</td> 
   <td>5 字节</td> 
   <td>长度小于等于 4294967295 的字节数组。</td> 
  </tr> 
 </tbody> 
</table>

如上表所示，三种长度的字节数组分别用不同长度的encoding字段来表示，用来节省空间。 而encoding的前两位用来标记encoding本身的类型。

##### 2、整数 #####

保存整数的时候，encoding字段为一字节长， 值的最高位以 11 开头。 整数值的类型和长度由编码除去最高两位之后的其他位记录。

<table> 
 <thead> 
  <tr> 
   <th>编码</th> 
   <th>编码长度</th> 
   <th>content 属性保存的值</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>11000000</td> 
   <td>1 字节</td> 
   <td>int16_t 类型的整数。</td> 
  </tr> 
  <tr> 
   <td>11010000</td> 
   <td>1 字节</td> 
   <td>int32_t 类型的整数。</td> 
  </tr> 
  <tr> 
   <td>11100000</td> 
   <td>1 字节</td> 
   <td>int64_t 类型的整数。</td> 
  </tr> 
  <tr> 
   <td>11110000</td> 
   <td>1 字节</td> 
   <td>24 位有符号整数。</td> 
  </tr> 
  <tr> 
   <td>11111110</td> 
   <td>1 字节</td> 
   <td>8 位有符号整数。</td> 
  </tr> 
  <tr> 
   <td>1111xxxx</td> 
   <td>1 字节</td> 
   <td>使用这一编码的节点没有相应的 content 属性， 因为编码本身的 xxxx 四个位已经保存了一个介于 0 和 12 之间的值， 所以它无须 content 属性。</td> 
  </tr> 
 </tbody> 
</table>

当encoding最前两位字段为11的时候，表示当前结点为整数。 同时encoding的后几位用来表示不同的整数类型。可以看到后几位中用000000表示int16\_t 类型的整数， 用010000表示int32\_t 类型的整数， 用100000表示int64\_t 类型的整数。

可以注意到，为了进一步节省内存，当编码为1111xxxx时，表示没有内容部分，xxxx已经存放了当前的整数值，包括整数0~12，即xxxx可以表示0000~1101。这样就节省了content的内存空间。这边编码为11111111代表ziplist的结尾。

### 连锁更新 ###

由于每个压缩列表的结点保存了上一个结点的大小，所以当前结点的变化有可能引起下一个结点的变化。如果前一节点的长度小于 254 字节， 那么 previous\_entry\_length 属性需要用 1 字节长的空间来保存这个长度值； 如果超过了254字节，这个属性值就需要 5 个字节的长度来保存。

所以最坏的情况下，压缩列表中某一个结点的更新，会引起所有结点的一个更新操作，就是所谓的连锁更新。

此外，插入或者删除结点也有可能引起连锁更新的操作。不过虽然连锁更新带来的消耗很大，但是仍旧可以放心的使用压缩列表，因为连锁更新引起的条件比较苛刻，概率比较小。 首先， 压缩列表里要恰好有多个连续的、长度介于 250 字节至 253 字节之间的节点， 连锁更新才有可能被引发， 在实际中， 这种情况并不多见；  
其次， 即使出现连锁更新， 但只要被更新的节点数量不多， 就不会对性能造成任何影响： 比如说， 对三五个节点进行连锁更新是绝对不会影响性能的。

## 压缩列表基本操作 ##

### 创建新的压缩列表 ###

/* Create a new empty ziplist. */
    unsigned char *ziplistNew(void) {
        // 空ziplist的大小为11个字节，头部10字节，尾部1字节
        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);
        // 压缩列表结点数为0
        ZIPLIST_LENGTH(zl) = 0;
        // 设定尾部一个字节位0xFF
        zl[bytes-1] = ZIP_END;
        return zl;
    }

### 插入结点 ###

由于连锁更新的存在，插入结点的复杂度平均 O(N) ，最坏 O(N^2)

// ziplist插入节点只能往头或者尾部插入
    // zl: 待插入的ziplist
    // s，slen: 待插入节点和其长度
    // where: 带插入的位置，0代表头部插入，1代表尾部插入
    unsigned char *ziplistPush(unsigned char *zl, unsigned char *s, unsigned int slen, int where) {
        unsigned char *p;
        // 获取插入的位置
        p = (where == ZIPLIST_HEAD) ? ZIPLIST_ENTRY_HEAD(zl) : ZIPLIST_ENTRY_END(zl);
        // 执行具体的插入过程
        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; /* 为了防止警告，进行初始化；用一个比较特殊的值以便能够方便的观察到不恰当的使用 */
        zlentry tail;
    
        /* Find out prevlen for the entry that is inserted. */
        if (p[0] != ZIP_END) {
            // 如果不是压缩列表的结束标志，说明p指向了一个已存在的结点
            // 解码得到p的前置结点和长度
            ZIP_DECODE_PREVLEN(p, prevlensize, prevlen);
        } else {
            // 如果p指向列表末端，表示列表为空
            unsigned char *ptail = ZIPLIST_ENTRY_TAIL(zl);
            if (ptail[0] != ZIP_END) {
                prevlen = zipRawEntryLength(ptail);
            }
        }
    
        /* See if the entry can be encoded */
        // 判断是否能够编码为整数
        if (zipTryEncoding(s,slen,&value,&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. */
        // 如果不是插入到列表的末端，都需要判断下一个结点是否能存放新节点的长度编码
        // 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;
        // 重新分配空间，curlen当前列表的长度
        // reqlen 新节点的全部长度
        // nextdiff 新节点的后继节点扩展header的长度
        zl = ziplistResize(zl,curlen+reqlen+nextdiff);
        // 根据新的压缩列表地址得到新的p的地址
        p = zl+offset;
    
        /* 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);
    
            /* 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. */
            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);
        }
    
        // 如果nextdiff不等于0， 下一个结点的头部需要进行扩展
        if (nextdiff != 0) {
            offset = p-zl;
            zl = __ziplistCascadeUpdate(zl,p+reqlen);
            p = zl+offset;
        }
    
        /* Write the entry */
        // 将新节点前置节点的长度写入新节点的header
        p += zipStorePrevEntryLength(p,prevlen);
        // 编码新结点
        p += zipStoreEntryEncoding(p,encoding,slen);
        if (ZIP_IS_STR(encoding)) {
            memcpy(p,s,slen);
        } else {
            zipSaveInteger(p,value,encoding);
        }
        ZIPLIST_INCR_LENGTH(zl,1);
        return zl;
    }

### 查找结点 ###

/* 寻找节点值和 vstr 相等的列表节点，并返回该节点的指针。 * 每次比对之前都跳过 skip 个节点。 * 如果找不到相应的节点，则返回 NULL 。 */
    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) {
            unsigned int prevlensize, encoding, lensize, len;
            unsigned char *q;
    
            // 解码得到前置结点的长度
            ZIP_DECODE_PREVLENSIZE(p, prevlensize);
            // 当前结点的长度
            ZIP_DECODE_LENGTH(p + prevlensize, encoding, lensize, len);
            q = p + prevlensize + lensize;
    
            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 {
                    /* 查看目标值是否能被编码，只会在第一次循环的时候检查； * 检查一次之后vencoding会被置为非0 */
                    if (vencoding == 0) {
                        if (!zipTryEncoding(vstr, vlen, &vll, &vencoding)) {
                            /* 如果不能被编码，设置格式为UCHAR_MAX ， 下次不会再检查*/
                            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) {
                        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;
    }

### 删除结点 ###

// 删除给定节点，输入压缩列表zl和指向删除节点的指针p
    unsigned char *ziplistDelete(unsigned char *zl, unsigned char **p) {
        size_t offset = *p-zl;
        // 调用底层函数__ziplistDelete进行删除操作
        zl = __ziplistDelete(zl,*p,1);
        // 删除操作可能会改变zl，因为会重新分配内存
        *p = zl+offset;
        return zl;
    }
    
    /* 从位置 p 开始，连续删除 num 个节点。 * 函数的返回值为处理删除操作之后的 ziplist */
    unsigned char *__ziplistDelete(unsigned char *zl, unsigned char *p, unsigned int num) {
        unsigned int i, totlen, deleted = 0;
        size_t offset;
        int nextdiff = 0;
        zlentry first, tail;
    
        zipEntry(p, &first);
        // 计算被删除节点的总个数
        for (i = 0; p[0] != ZIP_END && i < num; i++) {
            p += zipRawEntryLength(p);
            deleted++;
        }
    
        // totlen 是所有被删除节点总共占用的内存字节数
        totlen = p-first.p; /* Bytes taken by the element(s) to delete. */
        if (totlen > 0) {
            if (p[0] != ZIP_END) {
                // 不是尾结点，表示被删除节点之后仍然有节点存在
    
                // 因为位于被删除范围之后的第一个节点的 header 部分的大小
                // 可能容纳不了新的前置节点，所以需要计算新旧前置节点之间的字节数差
                nextdiff = zipPrevLenByteDiff(p,first.prevrawlen);
    
                /* Note that there is always space when p jumps backward: if * the new previous entry is large, one of the deleted elements * had a 5 bytes prevlen header, so there is for sure at least * 5 bytes free and we need just 4. */
                // 如果有需要的话，将指针 p 后退 nextdiff 字节，为新 header 空出空间
                // 由于会删除之前的结点，所以肯定会有足够的空间用来扩展
                p -= nextdiff;
                // 将 first 的前置节点的长度编码至 p 中
                zipStorePrevEntryLength(p,first.prevrawlen);
    
                /* Update offset for tail */
                ZIPLIST_TAIL_OFFSET(zl) =
                    intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))-totlen);
    
                /* 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, &tail);
                if (p[tail.headersize+tail.len] != ZIP_END) {
                    ZIPLIST_TAIL_OFFSET(zl) =
                       intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
                }
    
                /* Move tail to the front of the ziplist */
                // 从表尾向表头移动数据，覆盖被删除节点的数据
                memmove(first.p,p,
                    intrev32ifbe(ZIPLIST_BYTES(zl))-(p-zl)-1);
            } else {
                /* The entire tail was deleted. No need to move memory. */
                // 执行这里，表示被删除节点之后已经没有其他节点了， 不需要移动结点
                ZIPLIST_TAIL_OFFSET(zl) =
                    intrev32ifbe((first.p-zl)-first.prevrawlen);
            }
    
            /* Resize and update length */
            // 缩小并更新 ziplist 的长度
            offset = first.p-zl;
            zl = ziplistResize(zl, intrev32ifbe(ZIPLIST_BYTES(zl))-totlen+nextdiff);
            ZIPLIST_INCR_LENGTH(zl,-deleted);
            p = zl+offset;
    
            /* When nextdiff != 0, the raw length of the next entry has changed, so * we need to cascade the update throughout the ziplist */
            // 如果 p 所指向的节点的大小已经变更，那么进行级联更新
            // 检查 p 之后的所有节点是否符合 ziplist 的编码要求
            if (nextdiff != 0)
                zl = __ziplistCascadeUpdate(zl,p);
        }
        return zl;
    }

[SouthEast]: /images/20220608/380983d9e0314ab8ac67af564e66ffa1.png